feat(migrate): wandb → Pluto historical data migration - #131
Conversation
… disable_system_metrics Groundwork for pluto.migrate (wandb importer): explicit historical timestamps thread through to the sync layer (wire format already carried them), console lines can be replayed with original times, and the importing host's system metrics can be suppressed so they don't pollute migrated runs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Long-schema PartWriter with size-based part rotation, part readers, atomic-JSON state helpers, export sentinels, and the load-phase LoadedCache. wandb+pyarrow become the optional 'migrate' extra. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Full-fidelity export via wandb.Api: scan_history metrics with original step/timestamp, media/histogram rows, events-stream system metrics (renamed system.* -> sys/*), console lines from output.log (parsing per-line timestamps when present), artifacts with a size cap, and atomic per-run staging with sentinel-based resume. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Per-step metric replay with original timestamps, run createdAt via settings.compat, media/table/histogram conversion, console + artifact replay, sync-queue backpressure, finish-code mapping, external-id dedup, loaded-cache resume, and dry-run. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Thin argparse layer over WandbExporter/PlutoLoader; heavy deps import inside handlers so the base CLI works without the migrate extra. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Loads a hand-staged export into the dev/staging environment via PlutoLoader and verifies through pluto.query that historical metric timestamps, tags, and (once the server fix deploys) run createdAt round-trip. Gated on PLUTO_STAGING_API_KEY; URLs default to the pluto-*-dev.trainy.ai channel and are overridable via env. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
- perf-mode log() now carries timestamp through the queue (was silently dropped when sync is disabled) - collision on externalId resumes and re-replays instead of permanently marking a half-loaded run as done (typed RunExistsError in op.py) - metric replay batches groups through one SQLite transaction (Op._log_metrics_batch / SyncProcessManager.enqueue_metrics_batch) - disable_system_metrics now reaches the sync subprocess (sys/pluto.* health metrics no longer stamp migrated runs with current time) and suppresses host systemMetadata on run creation - backpressure wait is bounded (stall_timeout) with guarded polling - staged system metrics keep source-native names; loader owns the sys/ translation; console lines are no longer rewritten when they carry their own timestamps - 'all' loads staged runs even when some exports failed; 'all --dry-run' is rejected instead of silently exporting; --artifact-max-size-mb 0 means a zero cap, not unlimited Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a robust two-phase migration pipeline (pluto.migrate) to export historical experiment data from Weights & Biases to local parquet files and load them into Pluto while preserving original timestamps. The changes include new CLI commands, exporter and loader modules, resume bookkeeping, and support for historical timestamps and batch logging in the core Op class. The code review identified critical path traversal vulnerabilities when handling externally-sourced identifiers (such as run IDs, file names, and artifact names) from the wandb API and staged files. Additionally, several bugs and robustness issues were highlighted, including a missing defaultdict import in pluto/op.py, potential crashes from malformed JSON or non-string inputs, uncleaned temporary directories on export failure, and platform-dependent file opening without explicit UTF-8 encoding.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| new_metric_names: List[str] = [] | ||
| new_file_meta: Dict[str, List[str]] = defaultdict(list) |
There was a problem hiding this comment.
The defaultdict class is used here but it is not imported in this file, which will raise a NameError when _log_metrics_batch is called. Import defaultdict from collections locally or at the top of the file to prevent this crash.
| new_metric_names: List[str] = [] | |
| new_file_meta: Dict[str, List[str]] = defaultdict(list) | |
| from collections import defaultdict | |
| new_metric_names: List[str] = [] | |
| new_file_meta: Dict[str, List[str]] = defaultdict(list) |
| if self.before_ms is not None and created_ms > self.before_ms: | ||
| continue | ||
|
|
||
| run_dir = runs_root / run.id |
There was a problem hiding this comment.
The run.id is an externally-sourced identifier from the wandb API. Using it directly to construct run_dir can lead to path traversal vulnerabilities if a malicious run ID contains .. or path separators. Sanitize it using Path(run.id).name and check for unsafe values like . or .. before constructing the path.
| run_dir = runs_root / run.id | |
| safe_run_id = Path(run.id).name | |
| if not safe_run_id or safe_run_id in ('.', '..'): | |
| logger.warning(f'{tag}: unsafe run ID {run.id!r}, skipping') | |
| continue | |
| run_dir = runs_root / safe_run_id |
References
- Sanitize externally-sourced identifiers before using them to construct file paths to prevent path traversal vulnerabilities. Use
pathlib.Path(identifier).nameoros.path.basename(identifier)and check for unsafe values like.or...
| for f in run.files(): | ||
| if not self.include_files and f.name != 'output.log': | ||
| continue | ||
| try: | ||
| f.download(root=str(files_dir), exist_ok=True) |
There was a problem hiding this comment.
The f.name is an externally-sourced file path from the wandb API. To prevent path traversal vulnerabilities, sanitize it using Path(f.name).name and check for unsafe values like . or .. before downloading.
for f in run.files():
if not self.include_files and f.name != 'output.log':
continue
safe_name = Path(f.name).name
if not safe_name or safe_name in ('.', '..') or safe_name != f.name:
logger.warning(f'{tag}: unsafe file name {f.name!r}, skipping')
continue
try:
f.download(root=str(files_dir), exist_ok=True)References
- Sanitize externally-sourced identifiers before using them to construct file paths to prevent path traversal vulnerabilities. Use
pathlib.Path(identifier).nameoros.path.basename(identifier)and check for unsafe values like.or...
| f'({size} bytes > cap {self.artifact_max_bytes})' | ||
| ) | ||
| continue | ||
| dest = tmp_dir / 'artifacts' / artifact.name |
There was a problem hiding this comment.
The artifact.name is an externally-sourced identifier from the wandb API. To prevent path traversal vulnerabilities, sanitize it using Path(artifact.name).name and check for unsafe values like . or .. before constructing the download destination path.
safe_name = Path(artifact.name).name
if not safe_name or safe_name in ('.', '..'):
logger.warning(f'{tag}: unsafe artifact name {artifact.name!r}, skipping')
continue
dest = tmp_dir / 'artifacts' / safe_nameReferences
- Sanitize externally-sourced identifiers before using them to construct file paths to prevent path traversal vulnerabilities. Use
pathlib.Path(identifier).nameoros.path.basename(identifier)and check for unsafe values like.or...
| path = run_dir / (row['file_value'] or '') | ||
| if not row['file_value'] or not path.exists(): |
There was a problem hiding this comment.
The row['file_value'] is read from the staged parquet files, which could be manipulated or contain unsafe paths. To prevent path traversal vulnerabilities, sanitize it using Path(file_value).name and check for unsafe values like . or .. before checking for its existence or loading it.
| path = run_dir / (row['file_value'] or '') | |
| if not row['file_value'] or not path.exists(): | |
| file_value = row['file_value'] or '' | |
| safe_file_value = Path(file_value).name | |
| if not safe_file_value or safe_file_value in ('.', '..') or safe_file_value != file_value: | |
| logger.warning(f'{tag}: unsafe file path {file_value!r}, skipping') | |
| return | |
| path = run_dir / safe_file_value | |
| if not path.exists(): |
References
- Sanitize externally-sourced identifiers before using them to construct file paths to prevent path traversal vulnerabilities. Use
pathlib.Path(identifier).nameoros.path.basename(identifier)and check for unsafe values like.or...
| def parse_iso_ms(value: Optional[str]) -> Optional[int]: | ||
| """Parse an ISO-8601 string to epoch milliseconds (UTC assumed if naive).""" | ||
| if not value: | ||
| return None |
There was a problem hiding this comment.
The parse_iso_ms function assumes value is always a string. If the wandb API returns a non-string type (such as an integer or a datetime object), calling .replace will raise an AttributeError and crash the export. Add a defensive check to ensure value is a string.
| def parse_iso_ms(value: Optional[str]) -> Optional[int]: | |
| """Parse an ISO-8601 string to epoch milliseconds (UTC assumed if naive).""" | |
| if not value: | |
| return None | |
| def parse_iso_ms(value: Optional[str]) -> Optional[int]: | |
| """Parse an ISO-8601 string to epoch milliseconds (UTC assumed if naive).""" | |
| if not isinstance(value, str): | |
| return None | |
| try: |
| def _export_run(self, run: Any, run_dir: Path) -> None: | ||
| tmp_dir = run_dir.with_name(run_dir.name + '.tmp') | ||
| if tmp_dir.exists(): | ||
| shutil.rmtree(tmp_dir) # leftovers from an interrupted export | ||
| tmp_dir.mkdir(parents=True) |
There was a problem hiding this comment.
If an exception occurs during _export_run (e.g., during history scanning or file downloading), the partial .tmp directory is left on disk. Wrap the export logic in a try-except block to clean up tmp_dir on failure so orphaned temporary directories do not accumulate.
def _export_run(self, run: Any, run_dir: Path) -> None:
tmp_dir = run_dir.with_name(run_dir.name + '.tmp')
if tmp_dir.exists():
shutil.rmtree(tmp_dir) # leftovers from an interrupted export
tmp_dir.mkdir(parents=True)
try:
created_ms = parse_iso_ms(getattr(run, 'created_at', None))
with PartWriter(tmp_dir) as writer:
self._write_run_json(run, tmp_dir, created_ms)
self._export_history(run, writer)
if self.include_system:
self._export_system_metrics(run, writer)
files_dir = tmp_dir / 'files'
if self.include_files or self.include_console:
self._download_files(run, files_dir)
if self.include_console:
self._export_console(run, writer, files_dir, created_ms)
if self.include_artifacts:
self._export_artifacts(run, writer, tmp_dir)
mark_run_exported(tmp_dir, {'rows': writer.rows_written})
if run_dir.exists():
shutil.rmtree(run_dir)
os.rename(tmp_dir, run_dir)
except Exception:
if tmp_dir.exists():
shutil.rmtree(tmp_dir)
raise| return | ||
| base = self._row_base(run) | ||
| fallback_ms = created_ms or 0 | ||
| with open(output_log, errors='replace') as f: |
There was a problem hiding this comment.
Opening files without specifying an explicit encoding is platform-dependent and can lead to UnicodeDecodeError or corrupt characters on platforms where the default encoding is not UTF-8 (such as Windows). Specify encoding='utf-8' explicitly.
| with open(output_log, errors='replace') as f: | |
| with open(output_log, encoding='utf-8', errors='replace') as f: |
| def write_json_atomic(path: Union[str, Path], obj: Any) -> None: | ||
| path = Path(path) | ||
| tmp = path.with_name(path.name + '.tmp') | ||
| with open(tmp, 'w') as f: |
There was a problem hiding this comment.
Opening files without specifying an explicit encoding is platform-dependent and can lead to UnicodeDecodeError or corrupt characters on platforms where the default encoding is not UTF-8 (such as Windows). Specify encoding='utf-8' explicitly.
| with open(tmp, 'w') as f: | |
| with open(tmp, 'w', encoding='utf-8') as f: |
|
|
||
|
|
||
| def read_json(path: Union[str, Path]) -> Any: | ||
| with open(path) as f: |
There was a problem hiding this comment.
Opening files without specifying an explicit encoding is platform-dependent and can lead to UnicodeDecodeError or corrupt characters on platforms where the default encoding is not UTF-8 (such as Windows). Specify encoding='utf-8' explicitly.
| with open(path) as f: | |
| with open(path, encoding='utf-8') as f: |
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ity guard)
pluto.log({"phase": "warmup"}) now works natively, mirroring the wandb
migration: a bare string value is routed to the string-series data path
(mlop_data, dataType="string-series", raw value, logName registered as DATA)
instead of being silently dropped. Numeric/media/data paths are unchanged.
- op.py: route str values in _process_log_item_sync; register string keys under
the DATA log type; new _enqueue_string_series_sync helper.
- sync/process.py: send string-series `data` raw (not JSON-wrapped), matching
the migration wire and the reader; widen enqueue_data's data_dict type.
No cardinality/distinct-value guard: every string value is kept (no data loss),
regardless of how many distinct values the series has. The only value not sent
is a single point over 200 chars (a stray blob, not a state label), warned once
per key. The migration loader's cardinality guard is likewise removed.
Verified end-to-end against a local stack: phase -> string-series timeline,
an all-distinct 12-point series keeps all 12 points.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Adds a native sweep API mirroring wandb's, single-machine (no launcher):
- pluto.sweep(config) -> id: validate a wandb-shaped search space
({method, metric, parameters}); grid + random supported (bayes rejected with
a clear message, planned via optuna). Stored in-process + on disk.
- pluto.agent(sweep_id, fn, count): client-side "brain" enumerates (grid) or
samples (random) the space and runs fn once per combination. The sampled
hyperparameters are injected into each run's config and the run is tagged
sweep:<id> (init.py hook), so runs group under their sweep — exactly the data
model a sweep dashboard needs. Auto-finishes runs the fn leaves open.
Migration: the exporter now captures a run's wandb sweep (id/name/search-space
config) into the manifest instead of flagging it dropped; the loader tags the
run sweep:<id> and stores the sweep under config.wandb.sweep. Native and
migrated sweeps converge on the same tag+config model.
Verified end-to-end against a local stack: a 2x2 grid produces 4 tagged runs
with their combos; a real wandb sweep run migrates tagged with its search space.
Adds tests/test_sweep.py; migrate tests updated.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Resume (grid + random), no backend endpoint needed — the SDK lists a sweep's COMPLETED runs via pluto.query.list_runs(tags=["sweep:<id>"]): - grid: skip combinations whose run already completed. - random: run count - (already done). Best-effort: if the query fails (offline / project not created), the agent just runs everything, so a fresh sweep is never blocked. Bayesian search (method="bayes") via optuna (optional dep, pluto[sweep]): - agent asks optuna for each next combination, runs it, and tells optuna the objective (the run's final metric value), learning as it goes. - direction from metric.goal (minimize/maximize); requires metric + count. - resume seeds the study from completed runs' (params -> objective) so a restarted bayes search keeps learning; also caps at the total count. Supporting change: Op caches the latest numeric value per key (op._latest_metrics), and pluto.init hands the sweep run's Op back to the agent, so the objective is read directly (pluto.ops is unreliable here — finish() mutates it and object ids get reused). Verified end-to-end on a local stack: grid resume runs only the remaining combinations; a bayes search over (x-0.7)^2 converges near x=0.7. Adds bayes + resume tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
A native pluto.sweep() previously put only the sampled combo + the sweep:<id>
tag on each run, so the server never saw the sweep's method / objective / search
space and had to infer them. The agent now also stamps the declared spec onto
each run's config as `config.sweep` = {id, method, metric, parameters} — the
same shape migrated sweeps carry in `config.wandb.sweep`. The sweep dashboard
can now read the real declaration (correct optimize goal, the actual search
space, grid/random/bayes) instead of guessing, without any backend endpoint.
Implementation: agent() sets a module-level `_active_declared` for the run and
init() stamps it alongside the combo; cleared in a finally (covers the bayes
early-return). Adds a test.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…tion
Adds an `annotations` field to the file-upload path (mirrors `caption`) that
carries a wandb-shape {boxes, masks} JSON string, forwarded verbatim to the
server (mlop_files.annotations) as the render side expects.
- pluto.Image now takes `boxes` (native, {layer: {box_data, class_labels}}) and
`annotations` (a ready JSON blob, used by migration). Boxes default to
domain="pixel" so coords aren't misread as 0-1 fractions.
- Plumbing: File._annotations -> op._enqueue_file_sync -> SyncManager.enqueue_file
-> store (file_uploads.annotations column, v3 additive migration) ->
upload_files_batch (only sent when set, like caption).
- Migration: the exporter stages the wandb box/mask refs in a new parquet
`annotation_value` column (was: self._skipped('image-annotations')); the loader
resolves the .boxes2D.json sidecar into the image's annotations. Coverage now
reports `image-boxes` migrated.
Verified end-to-end into the sync DB for both native (boxes with domain=pixel)
and migration (real run's boxes resolved to {box_data, class_labels}); the field
name matches the ingest's serde `annotations` exactly.
Masks (a separate PNG to re-upload + hide from the media grid) are still
deferred — flagged `image-masks`. Adds exporter + loader tests.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Completes image annotations. Each mask layer becomes a PNG uploaded in the
image's log group with fileType "mask" (the convention the render side hides
from the media grid and resolves by fileName), referenced from the image's
annotations as {fileName, class_labels?}.
- pluto.Image now takes `masks`: native {layer: {mask_data: <HxW class-id
array>, class_labels}} is encoded to a PNG with the class id in the red
channel (what the renderer reads); migration {layer: {path}} forwards wandb's
own mask.png. Mask PNGs to upload alongside are collected in
`_annotation_files`.
- op._enqueue_file_sync uploads those sub-files in the same log group and honors
File._upload_file_type; the sync payload sends fileType "mask" for them
(reuses the file_type field — no new column).
- Migration: masks now migrate (was flagged image-masks); the loader resolves
the staged .mask.png ref to a path and hands it to pluto.Image.
Verified end-to-end into the sync DB: native (numpy → red-channel PNG) and
migration (real run) both produce an image row with boxes+masks annotations and
a sibling mask row with fileType=mask whose fileName matches the reference.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The sweep commits added optuna to pyproject.toml (optional dep, dev dep, and the [sweep] extra) but never regenerated poetry.lock, leaving a stale content-hash. CI's `poetry install` refuses an out-of-sync lock, so every job failed at the "Install dependencies" step (format, mypy, tests, api-docs, contract-test — all 9). Re-locked with poetry 2.1.1 (matches CI). Adds only optuna + its transitive deps (alembic, colorlog, greenlet, mako, sqlalchemy); no other package versions change. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
wandb keeps two copies of a media table; the exporter stages the lossy run-files copy, so image cells load as the literal text "Image" rather than pictures. Pin that current behavior: the table migrates (not dropped, no crash) and the cell images arrive as a separate, unlinked artifact. This test should flip to assert real image cells when we wire cell refs to uploaded images. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…annotations Two regressions from the sweep and image-annotation commits, surfaced once CI's install step started passing: - _log_via_sync() unconditionally did self._latest_metrics.update(metrics), but call sites that build a bare Op via Op.__new__ (log-resilience unit tests) skip __init__ where it's initialized -> AttributeError. Logging is best-effort and must never crash, so lazy-init the cache at the use site (the sweep reader already uses getattr defensively). - pluto.Image gained boxes/masks/annotations and the public API gained sweep/agent, leaving docs-api/media.mdx and meta.json stale (api-docs --check failed). Regenerated with griffe 2.0.2 (matches the lock). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
A wandb.Table with media columns migrates degraded: we stage wandb's lossy run-files table copy (media cells are text placeholders like "Image"), and the cell media arrives only as the sibling run_table artifact, unlinked from the table. Previously this happened silently. The exporter now detects media columns by reading the downloaded table artifact's column_types for media *-file wb_types, and emits a table-media-cell coverage flag + warning (trips --strict). The table and its scalar columns still migrate; only the in-cell media is flagged as lost, mirroring the artifact-versioning partial-migration pattern. Verified against a real seeded media table (NOT migrated: 1 table-media-cell) and with unit tests for media vs plain tables. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…lags Add export-side coverage for the two skip paths not exercised by the seeded fixtures: the generic unsupported(<type>) branch (bokeh/joined- table/... media types the exporter can't stage) and string-series-too-long (a per-step string over the 200-char cap). FakeRun gains a history_rows override so a test can drive scan_history directly. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
wandb → Pluto migration — summary
The render half is server-private#559 — it draws what this PR imports (custom charts, media viewers, string metrics, sweeps, image overlays). Everything below is verified end-to-end against real migrated data. Native SDK users (
Commands
Knobs
What migratesEach row is the same seeded data in wandb and Pluto — screenshots, not links. Visual features are shown across multiple areas. Nothing crashes on any migrated run.
What does NOT fully migrateEach of these warns you during migration (and stops it in
Masks: one deployment caveat (CORS). Masks now render (see above — the exporter recovers Tables with images inside their cells migrate, but the images come loose. wandb keeps two copies of such a table: a full one (with real image links) and a stripped one where each image cell is just the word These aren't real gaps: Still to do: artifact version-history graph (frontend); a custom-chart dashboard widget (frontend — they already render on the run and all-runs pages); and hand-drawn custom charts ( Sweeps — how Pluto differs from wandbIn Pluto a sweep is just "all the runs tagged with that sweep's id" — no new database tables. Because of that, Pluto works out a sweep's state (running / finished / stopped early) from its runs, instead of storing one like wandb does. A few side effects: a crashed run keeps a sweep looking "running" until a background cleanup notices (~30 min); "stopped early" only applies to grid sweeps; and the run count is "how many were attempted" (a 6-run grid where 4 crashed still says "6/6", with the failures shown next to it). Parameter importance (which knob mattered most) uses the same math as wandb — the ranking matches, though the exact bar lengths won't, because wandb keeps its random seed secret. Not built: starting / pausing / stopping / resuming a sweep from the UI — all of which would need a live sweep controller running. BenchmarksA 740-run test project covering every data type and edge case (NaN/±Inf/null, failed & crashed runs, all sweep kinds, mixed-dtype and image-cell tables, boxes/masks, all artifact flavours, molecule/bokeh/joined-table, galleries, system metrics) was seeded to wandb, migrated into Pluto (with the #559 frontend), and checked three ways — timing, data loss (counted in the DB), and rendering.
Performance (
Resilience & recoveryInterruptions (network blips, redeploys, killed processes) are survivable — re-run the same command and it converges. Three mechanisms: crash-safe staging (parquet renamed only after a completion sentinel; load buffers to SQLite before upload), idempotent resume (
Caveats: a mid-upload interruption can duplicate some media on resume (metrics dedup by timestamp; loader warns); a corrupt wandb artifact (bad manifest) can't be healed by retry — flagged and skipped. Terminal status — always matches wandb (
Validated: a fresh 490-run migration hit one live false-reap, auto-healed it inline, and a status-by-status audit confirmed 490/490 correct, 0 stranded (+3% wall-clock for the read-backs). Recovery is client-side; preventing the reap is a separate server-side change, not required for correctness. Bugbot hardening ( |
… config) Migrated segmentation masks rendered blank because they arrived without class_labels (the id->name colour key). Root cause: wandb stores mask class_labels in the run *config* (_wandb.value["mask/class_labels"], keyed '<image>_wandb_delimeter_<layer>'), NOT in the mask media descriptor that scan_history returns — and separately, the loader was dropping class_labels from the staged mask ref (kept only the path). Fix, both on the pluto side (no server-private change needed — the #559 frontend already paints class_labels; native masks proved it): - exporter: recover class_labels from run config and fold them into each mask layer's annotation. - loader: carry class_labels through to the uploaded mask spec. Verified end-to-end: a wandb-imported mask now renders identically to a native pluto.Image(masks=). Adds exporter + loader unit tests; also lands a previously-pending artifact-download-failed export test. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The PR #131 summary now embeds screenshots via GitHub user-attachments, so the committed .github/pr131/*.png copies (and loose .github screenshots) are unused. Removes ~3.8 MB of tracked images. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
@cursor review |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
…led restores, attempt failed runs once Three defects in the load/all failure path (flagged in review of 0f61ca8): 1. loader: on a collision-restore where finish() throws, the run was still marked loaded and skipped — stranding it RUNNING server-side and skipping it forever. Now it's reported as failed (not marked loaded), so a later run retries it. 2/3. all: a run that failed one poll pass was re-attempted every subsequent pass (re-running identical staged data can't help and risks duplicating media on a mid-replay resume), and once added to the reported-failures set it could never be cleared. PlutoLoader now takes skip_run_ids; _all_one_project feeds already-failed run-ids into it, so each failure is at-most-once and the reported failure count stays exact. The loader only ever sees a run once its export sentinel is written, so a load failure is genuine (complete data) rather than "needs more time" — attempt-once is safe; the user re-runs all/load to retry (in-progress runs resume from the ledger). Adds regression tests: restore-failure is reported not marked loaded, skip_run_ids bypasses a run without attempting it, and all feeds failed run-ids into skip_run_ids on later passes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…on't leak processes The sync subprocess is a persistent daemon: its main loop only breaks on SIGTERM or when the parent process dies. Normal-mode stop(wait=True) flushed the run's data and returned without terminating it — so each run's subprocess survived until the *caller* exited. For a short-lived caller (a training script) that's harmless: the process ends and orphan-detection reaps the subprocess. But a long-lived caller that starts one sync process per run — a single `pluto migrate wandb load` invocation over a whole project — accumulates one live subprocess per run. A 740-run load piled up ~584 subprocesses and drove a 32 GB host to 2 GB free (near-OOM). The localhost benchmark only avoided this by splitting the load into many separate 50-run invocations, whose exits reaped each batch — masking the leak rather than fixing it. stop(wait=True) now terminates the subprocess once its data is flushed (SIGTERM -> bounded wait -> SIGKILL fallback), mirroring the wait=False branch. Each subprocess's lifetime is scoped to its run, so a whole-project single-invocation load stays flat (a 20-run load peaks at ~3 concurrent sync processes instead of 20). Adds a regression test asserting the subprocess is dead after finish(). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ated)
Images logged as a list (wandb.log({k: [img, img, ...]}), wandb type
images/separated) were migrating as plain pictures — their boxes/masks were
silently dropped. The exporter downloaded the mask files but wrote
annotation_value=None for every gallery image, so the loader never linked them.
Single images were fine; only the gallery/list path missed it: it never read the
per-image all_boxes[i] / all_masks[i] arrays. Extracted a shared
_image_annotation_value() helper (used by both the single-image and gallery
paths) that folds mask class_labels back in from the run config, and the gallery
path now emits per-image annotation_value.
Verified end-to-end: re-export of a gallery run stages 12/12 images with
annotation_value (was 0/12) — boxes + masks + class_labels — and a live
re-migration lands 240 mask files where there were 0. Adds a regression test.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ate drops A transient TCP reset on a run's final status POST (/api/runs/status/update) was swallowed as a "shutdown signal": _try returned None with no retry and a DEBUG log, update_status() ran with raise_on_error=False, and finish() never surfaced it. The migration loader then marked the run "loaded", so it stayed non-terminal server-side (UI renders it FAILED with an ever-growing duration) and a re-run skipped it — permanently stranded. Hit ~0.6% of a 490-run migration (3 runs), all reported as "0 failed". Three layers of defense: - iface: add retry_connection_errors so the terminal status POST retries dropped keep-alive sockets (with backoff) and raises on final failure, instead of the fire-and-forget shutdown shortcut (kept for heartbeat/trigger/upload spam). update_status() now runs raise_on_error=True + retry_connection_errors=True. - op: _teardown records an unconfirmed terminal status on op._status_update_error instead of swallowing it, and no longer re-flips an already-confirmed run to FAILED on a later teardown hiccup. - loader: don't mark_loaded a run whose finish wasn't confirmed — record it failed so the next all/load pass heals it via the RunExistsError->restore path. Tests: connection-reset retry/raise in _try + update_status wiring; loader routes an unconfirmed finish to failed (main + restore paths); finish() records the error without crashing and keeps a confirmed run COMPLETED. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…eaps A back-dated freshly-created import can be reaped to FAILED by the server's stale-run monitor in a sub-second race; the client's COMPLETED is then silently rejected (HTTP 200, no effect), so the run is stranded FAILED while the loader reports success. The earlier b1e1ddb only catches a finish that errors/drops — this one returns 200, so it slipped through. After a run finishes, read its status back (GET /api/runs/details/{id}) and, on a mismatch for a finished import, heal it by resume+finish — reopening the run to RUNNING and finishing again, which sidesteps the terminal-precedence guard the way a plain re-finish cannot, and replays nothing (no duplication). Bounded to 3 attempts; a still-unconfirmed run is recorded failed (a later pass retries) rather than cached as loaded. Unreadable status degrades to best-effort (don't fail a run over our own inability to verify). Scoped to the migration loader, so native runs are untouched. No server change; recovery only (it does not prevent the reap — the raceless prevention is the server-side lastActivityAt floor). Validated end-to-end: healed the one real stranded run (ca40i2mb / MFV-372) in migrate-final-v2 to COMPLETED. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
@cursor review |
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Comment @cursor review or bugbot run to trigger another review on this PR
Reviewed by Cursor Bugbot for commit 3000597. Configure here.
…ter, purge race - loader: `_verify_and_heal` now checks the heal's own finish (`op._status_update_error`) and no longer best-effort-marks a run loaded when a heal was left unconfirmed and the status can't be read back — it reports failed so a re-run retries (matches the main/restore paths). (High) - wandb_export: exclude runs with an unparseable `created_at` from a date-filtered export instead of silently including them past the requested window. (Medium) - wandb_export: the empty-cache purge only reaps 0-byte parquets older than `_EMPTY_CACHE_STALE_SECONDS`, so parallel projects sharing the wandb cache under `--workers>1` don't delete each other's in-progress (briefly 0-byte) downloads. (Medium) Tests: heal-unconfirmed -> failed (not loaded); undated run excluded when a date filter is set and kept when not; purge spares a fresh 0-byte file and reaps a stale one. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
screen-recording-2026-08-12-at-42317-pm_fsKYFg77.movwandb 9 custom charts vs pluto 10 custom charts: wandb 3 charts vs pluto 5 metrics: wandb 12 media vs pluto 76 media: wandb 13 table vs pluto 12 tables: wandb 10 sys vs pluto 12 sys: |
|
forgot video of sweep stuff: Screen.Recording.2026-08-12.at.5.00.12.PM.mov |
New native SDK additions in this PR (not in the docs yet)Besides the wandb migration, this PR adds three things you can log natively. Everything else in the docs — Setup: 1. String metrics —
|
2 participants
Summary
Two-phase migration tool for importing legacy wandb data into Pluto with full timestamp fidelity — built for the first customer coming off the wandb shim who wants their historical runs transferred.
pluto migrate wandb export— stages complete runs from the wandb cloud API on disk (parquet long-schema parts +run.jsonmanifest + media/artifact files): fullscan_historymetrics, media, tables/histograms, events-stream system metrics, console logs, artifacts (size-cappable). Atomic per-run staging with sentinel-based resume.pluto migrate wandb load— replays staged runs through the public client API with original wall-clock timestamps,wandb::{entity}/{project}/{run_id}external-id dedup, crash-healing resume (typedRunExistsError→ resume + re-replay), batched metric enqueue, bounded backpressure, and--dry-run.pluto migrate wandb all— both phases; loads whatever staged even if some exports failed.Client groundwork
op.log(..., timestamp=)— epoch-seconds override threaded through the sync store (wire format already carried per-point time); works in sync, legacy, and perf-queue modes.Op._log_console/Op._log_metrics_batch— batched backfill helpers.settings.disable_system_metrics— keeps the migration host's hardware/health metrics (sys/*,sys/pluto.*, runsystemMetadata) out of imported runs, including in the sync subprocess.migrateextra (pip install 'pluto-ml[migrate]'): wandb + pyarrow, lazily imported so the base CLI/package are unaffected.Server dependency
Run
createdAt/updatedAtbackfill needs the companion server PR (Trainy-ai/server-private branchfeat/run-createdat-backfill); metric/file/console point timestamps already round-trip against today's prod ingest. Until it deploys, imported runs show import-day creation dates only.Test plan
tests/test_migrate_staging_e2e.py: live round-trip against the dev channel, gated onPLUTO_STAGING_API_KEY.🤖 Generated with Claude Code
Note
High Risk
Large new migration and ingest surface (historical timestamps, run lifecycle, bulk replay, auth, and file uploads) with high blast radius if replay or status handling is wrong; server backfill for run
createdAt/statusUpdatedis a deployment dependency.Overview
Adds a wandb → Pluto migration pipeline (
pluto migrate wandb export | load | all) with on-disk parquet staging, resumable export/load ledgers, parallel project workers, and coverage reporting. The loader replays runs through the public client with historical wall-clock times,wandb::…external ids, terminal-status verification/healing, and optional staged-file cleanup.Backfill-oriented client changes:
log(..., timestamp=)(epoch seconds),Op._log_metrics_batch/_log_console,settings.disable_system_metrics, compatcreatedAt/updatedAt/systemMetadataandstatusUpdatedon finish, plusRunExistsErrorfor intentional resume. Terminal status updates retry connection resets and surface failure via_status_update_error. Login no longer overwrites an explicitly provided API key on a failed validation POST.Media:
Imagesupports boxes, masks, and annotations (mask PNGs upload asfileTypemask); sync/file upload carriesannotations.Also ships client-side hyperparameter sweeps (
pluto.sweep/pluto.agent, grid/random/bayes via optuna) withinit()merging sampled config andsweep:<id>tags—aligned with migrated sweep metadata.Sync subprocess is terminated after drain so long bulk loads do not accumulate daemons; string metrics route to string-series ingest. Docs/API meta updated for
timestamp,Imageparams, and sweep symbols. Requirespluto-ml[migrate]extra for wandb/pyarrow (lazy-loaded from CLI).Reviewed by Cursor Bugbot for commit 3000597. Configure here.
Summary by CodeRabbit
pluto migrate wandbCLI withexport,load, andallworkflows, resumable staging/loading, project scoping, and configurable parallel workers.timestampsupport forlog(...)to improve historical replay fidelity.disable_system_metricsto suppress host hardware/system metrics during backfills and migrations.statusUpdatedtiming.log(...)documentation to explaintimestampsemantics and invalid-value handling.