Fan-out/fan-in fleet: parallel-sweep engine recovered from spark, integrated with the MLE-STAR flow, proven live by cgpadwick · Pull Request #37 · cgpadwick/saage

cgpadwick · 2026-07-07T04:20:33Z

Stacked on #36 — review/merge that first.

What this is

The parallel-sweep line of work (26 commits built and live-tested on the spark box in June, never pushed) — recovered, pushed as parallel-sweep, and ported commit-group by commit-group onto the campaign branch, reconciled with everything that changed since the fork. Suite 431 → 512 tests. Each port commit credits the original spark commit.

The machinery

Capacity (H1/H4): max_runs slots per target; preflight rejects at capacity; free_slots()
Programmatic API (H3): poll_run (mirror-first) / fetch_run (reconciled with fetch --workspace) / kill_run / bucket_names
Provisioning (H6): content-keyed, flock-guarded, stamped node setup into a shared $SAAGE_CACHE; concurrent bootstraps populate the cache once
dispatch_many (H2): fan N jobs over K boxes — quarantine, per-job deadlines, dead-box detection (stale heartbeats AND ssh unreachable), lost-job requeue, stranded accounting; side effects behind an Ops seam, fully offline-tested
Batch rounds: K proposals → K parallel experiment runs → winner's patch, invoked by a coordinator flow as an ordinary command step — the engine never learns what "parallel" means
Fleet: sweep-up / sweep-watch / sweep-down — coordinator-on-a-box with per-sweep ssh keys + scoped credentials (cloud API keys never leave the laptop); workers auto-released at the flow's batch-done marker. Thunder backend rebuilt on the REST thunder_api (hairpin-NAT avoidance and (ip,port)-teardown preserved)
Flows: fmnist_batch (the phase-2 proving flow) and flow_batch.yaml — the batched kaggle solver modernized onto the MLE-STAR flow (researcher idea menu, memory, multi-seed baselines, self-ensemble, timeouts)

Proven live (two 5-box sweeps, spooky-author)

Sweep 1: rounds perfect (0.5549 → 0.3663 val, ~21 min/round ≈ 3.5× sequential throughput, $0.89 LLM for 30 experiments) — tail broke, exposing three real bugs
Fixes (each reproduced in a test first): retry/polling checkers treat a non-answer as fail (was: silent success); integrate --smoke-cmd — a winning patch must execute on the coordinator before acceptance (worker/coordinator package skew killed a final train); batch/sequential ledger schemas interleave
Sweep 2 (fixes live): 12 rounds, val 0.3973, test 0.4188 above median, $0.86 LLM, workers auto-released, full clean auto-teardown — zero manual touches

🤖 Generated with Claude Code

Ported from the parallel-sweep branch (built + live-tested on the spark box in June, never pushed). A target may host max_runs concurrent runs; preflight rejects at capacity instead of at one, and free_slots() gives schedulers the live headroom. Default stays 1 — existing targets are unchanged. Original work: spark:~/code/saage-sweep parallel-sweep (3edb2fb). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

poll_run (mirror-first status, one poll, no printing), fetch_run (node-first with mirror fallback — now carrying the --workspace pull that master grew in the meantime), kill_run, and bucket_names (cheap LIST of a run's mirror). The CLI verbs wrap these; a dispatcher builds on them without shelling out. Original work: spark parallel-sweep (ec8657f), reconciled with the fetch --workspace feature added on master since the fork. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

provision_node runs a setup command once per node under an flock, keyed by the command's hash, stamped on success — safe to re-run, and a replacement box just provisions cold. Bootstrap exports SAAGE_CACHE and serializes ws_setup hooks under a content-keyed lock, so K concurrent bootstraps on one box populate the dataset/venv cache once instead of racing K pulls. Original work: spark parallel-sweep (f42e8fa). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

A caller may name the run up front (the sweep launcher needs the node-side run dir path before handoff, for scoped credentials) and push extra 0600 files into the run dir (e.g. a coordinator's scoped saage_home). Paths are validated run-dir-relative — they land in generated shell. Original work: spark parallel-sweep (889c7e2, split). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…se 2) The heart of the parallel sweep, ported intact with its tests: - dispatch.py: fan N jobs over K boxes — slot accounting against max_runs, one-time provisioning, concurrent handoffs in a small thread pool, mirror-first polling, per-job deadlines, dead-box detection (stale heartbeats AND ssh unreachable), lost-job requeue, target quarantine, stranded-job accounting. Side effects behind an Ops seam so the whole loop is offline-testable. - batch.py: one batched hill-climb round — K proposals become K parallel experiment runs (each job stages its own flow-dir copy with proposal.md inside, ship-head workspace), scores collected from eval_results.json, winner picked direction-aware, verdict printed as BEST_INDEX=/BEST_SCORE=/BEST_PATCH= for a coordinator flow's set: captures. The engine never learns what 'parallel' means — a coordinator flow calls this as an ordinary command step. Original work: spark parallel-sweep (716f781, da51ab2, 24bf2b0). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…weep-down Ported from spark parallel-sweep (889c7e2, 718b484), with the Thunder backend rebuilt on saage.remote.thunder_api (REST) instead of the tnr CLI — same hard-won semantics preserved: hairpin-NAT avoidance (a worker must not share the coordinator's proxy IP), teardown matched on (ip, port) never IP alone, per-instance keys persisted at spawn. The security model travels intact: a per-sweep keypair authorized only on that sweep's workers; the coordinator gets a scoped saage_home (workers + storage mirror only) pushed 0600 into its run dir; cloud API keys never leave the laptop, so teardown stays laptop-side (sweep-watch releases workers at the flow's batch-done marker, then fetches from the mirror and tears everything down at the final phase). CLI: provision / sweep-up / sweep-watch / sweep-down subcommands, --slots on add-target and both spawn providers. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…first user Coordinator flow: propose_batch (K diverse proposals per round, critic- gated) -> one command step running saage.remote.batch (K parallel experiment runs across worker targets) -> integrate.py applies the winning patch and commits. Includes the standalone experiment.yaml each worker runs, the seeded FashionMNIST workspace, and the offline integration test of integrate.py's apply/commit/reject paths. Original work: spark parallel-sweep (a1b2b4f, e4712d7, 8b7b435). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…LE-STAR flow flow_batch.yaml — the P3 fan-out/fan-in variant, ported from spark parallel-sweep (a5294ac, 06602bc…de6c8ef) and modernized onto everything flow.yaml gained since the fork: - kept from the original: the pre-run researcher (ranked idea menu; format gated deterministically by check_ideas.py, substance gated by research_critic), K=3 diverse proposals per round via propose_batch, run_batch as an ordinary command step calling saage.remote.batch, integrate_batch.py applying the winning patch direction-aware, per-node one-time data provisioning, the batch_done.marker that lets sweep-watch release worker boxes early, and the worker-side experiment.yaml (proposal staged as flow data; experiment.patch + eval_results.json travel back as artifacts) - gained from this week's flow: memory staging, multi-candidate baseline seeding (--baseline auto), data_audit, train timeouts, v4-pro brains with flash critics, the self-ensemble stage with its deterministic keep-gate, the LLM report agent, and the fixed grade step with medal/test_score/above_median captures Their live results with the old base: best-ever 0.3745 val on spooky (researcher-menu run, June). This port gives the same machinery the flow that graded 0.3481 test. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Batched flows call back into the engine from a command step (`{{ engine_python }} -m saage.remote.batch`); sys.executable is the one interpreter guaranteed to have saage on its path, both locally and inside a remote node's per-run venv. Original work: spark parallel-sweep (da51ab2, split). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…heckpoints) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…or handoff The coordinator box is fresh — the kaggle flow needs cloud_setup.sh (torch/mlebench/data) exactly like a plain handoff does; and a sweep should be able to pin one model for every coordinator step. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…nv-skew guard, ledger schema Sweep 2003 postmortem (rounds worked perfectly: 10 rounds, 3 parallel experiments each, ~21min/round, val 0.5549 -> 0.3663; the TAIL broke): 1. engine: a retry_loop/polling_loop checker that answers neither pass nor fail silently ended the loop as success — the crashed final train slipped through when its verifier emitted no ACTION. A non-answer now routes to the guard (retry / keep polling). 2. flow: the round winner was validated on a WORKER's package set and crashed on the coordinator's (sklearn removed multi_class= between the two). integrate_batch gains --smoke-cmd: the winning patch must EXECUTE locally (1-epoch train) before it is accepted; a smoke failure reverts the patch and fails the round. Round artifacts and the researcher menu are now git-excluded so the revert's clean can't eat them. 3. helpers: batched rounds write round/slot ledger rows; the sequential keep_or_revert KeyError'd on them (the ensemble gate crashed post-final-train). Both writers now interleave cleanly — integrate rows carry step numbers, the reader tolerates rows without them. All three reproduced in tests before fixing. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Sweep 2 (all fixes live): 12 rounds x 3 parallel experiments over 4 workers, ~21 min/round, val 0.5629 -> 0.3973, test 0.4188 above median, $0.86 LLM, workers auto-released at the marker, clean auto-teardown. The round-4 breakthrough after three straight misses shows max_failed_rounds=4 is load-bearing. Sweep 1's row records the honest postmortem (rounds perfect, tail broke, three bugs -> fixed -> tested). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Copilot

Pull request overview

This PR recovers and integrates a “fan-out / fan-in” parallel-sweep system into saage, adding a slot-aware remote dispatcher, content-keyed provisioning, and coordinator-on-a-box fleet tooling, then wiring it into batched hill-climb flows (notably the Kaggle MLE-STAR solver and an FMNIST reference flow) with extensive offline test coverage and documentation.

Changes:

Add remote parallel execution primitives: per-target capacity (max_runs), programmatic poll/fetch/kill API, content-keyed provision-once caching, and a resilient dispatch_many scheduler with quarantine/dead-box handling.
Add “batch round” orchestration (saage.remote.batch) and deterministic integrators for batched hill-climbs; introduce new batched flows for Kaggle solver and FMNIST reference.
Harden engine control-flow primitives so retry/poll loops treat non-answers as non-success, and document the overall batched hill-climb plan and usage.

Reviewed changes

Copilot reviewed 49 out of 50 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
tests/test_kaggle_batch_helpers.py	New unit tests for Kaggle batched integration, worker setup, and idea-menu checks.
tests/test_fmnist_integrate.py	New unit tests for FMNIST batched round integrator behavior.
tests/remote/test_scripts.py	Updates bootstrap script expectations for cache + locked ws-setup.
tests/remote/test_provision.py	New tests for provision stamping/locking/idempotence semantics.
tests/remote/test_fleet.py	New offline tests for sweep credential scoping and teardown behavior.
tests/remote/test_dispatch.py	New offline tests for dispatcher scheduling, retries, quarantine, and dead-box logic.
tests/remote/test_capacity.py	New tests for `max_runs` capacity and preflight guards.
tests/remote/test_batch.py	New offline tests for phase-2 batch round staging/selection behavior.
tests/remote/test_api.py	New tests for programmatic poll/fetch/kill surfaces and mirror-first behavior.
tests/integration/test_research_loop.py	New integration test for nested researcher format/substance gating behavior.
saage/remote/target.py	Adds capacity enforcement in preflight and `free_slots()` utility.
saage/remote/scripts.py	Adds shared cache export + ws-setup serialization via content-keyed flock.
saage/remote/provision.py	New content-keyed, locked, stamped node provisioning helper.
saage/remote/observe.py	Adds `bucket_names()` and programmatic `poll_run`/`fetch_run`/`kill_run` APIs; refactors CLI to use them.
saage/remote/handoff.py	Adds `run_id` override and `push_files` support with path validation.
saage/remote/fleet.py	New sweep-up/watch/down coordinator-on-a-box fleet tooling with scoped creds.
saage/remote/dispatch.py	New dispatcher implementation (`dispatch_many`) with slot accounting and reaper logic.
saage/remote/creds.py	Adds `max_runs` to target configuration and persistence.
saage/remote/cli.py	Adds CLI support for slots, provision, and sweep commands; expands error handling.
saage/remote/batch.py	New batch-round runner bridging sequential flows to parallel dispatch.
saage/remote/init.py	Exposes new remote APIs (dispatch/provision/observe) at package level.
saage/primitives.py	Treats retry/polling loop non-answers as fail/running instead of silent success.
saage/hydrate.py	Seeds `engine_python` into the shared store for reliable `python -m saage...` callbacks.
README.md	Documents fan-out/fan-in batched parallel hill-climbs and new remote commands.
flows/kaggle_solver/worker_setup.py	New worker-side setup for batched experiments (excludes + data link from cache).
flows/kaggle_solver/setup_competition.py	Extends excludes to protect batch artifacts and idea menu across reverts.
flows/kaggle_solver/research_ideas/skill.md	New researcher skill producing structured ranked idea menus.
flows/kaggle_solver/research_critic/skill.md	New critic skill judging menu substance (post-format gate).
flows/kaggle_solver/README.md	Adds benchmark rows and narrative notes for batched sweep runs.
flows/kaggle_solver/propose_critic_batch/skill.md	New critic skill for 3-proposal batched rounds.
flows/kaggle_solver/propose_batch/skill.md	New proposer skill to emit 3 diverse proposals per round.
flows/kaggle_solver/memory/spooky-author-identification.md	Updates stored competition memory to reflect batched sweep run results.
flows/kaggle_solver/keep_or_revert.py	Makes sequential ledger writer tolerant of batch ledger rows lacking `step`.
flows/kaggle_solver/integrate_batch.py	New deterministic integrator for batched Kaggle rounds (direction-aware + smoke guard).
flows/kaggle_solver/flow_batch.yaml	New batched variant of the Kaggle solver flow using remote batch rounds.
flows/kaggle_solver/experiment.yaml	New worker-side mini-flow used as the per-proposal experiment job.
flows/kaggle_solver/check_ideas.py	New deterministic validator/renderer for researcher idea menus.
flows/kaggle_solver/BENCHMARK_RESULTS.md	Records new batched sweep benchmark entries.
flows/kaggle_solver/benchmark_journal.jsonl	Adds journal rows for the new batched sweep runs.
flows/fmnist_batch/seed/train.py	New baseline train contract for FMNIST batched reference flow.
flows/fmnist_batch/seed_workspace.py	New deterministic workspace seeding script for FMNIST coordinator.
flows/fmnist_batch/README.md	New documentation for the FMNIST batch reference flow and live results.
flows/fmnist_batch/propose_critic/skill.md	New critic skill for FMNIST proposal-set gating.
flows/fmnist_batch/propose_batch/skill.md	New proposer skill to emit 3 diverse FMNIST proposals per round.
flows/fmnist_batch/integrate.py	New deterministic FMNIST batch round integrator.
flows/fmnist_batch/implement_experiment/skill.md	New implementation skill for FMNIST worker experiments.
flows/fmnist_batch/flow.yaml	New FMNIST batch coordinator flow using `saage.remote.batch`.
flows/fmnist_batch/experiment.yaml	New FMNIST worker mini-flow generating `eval_results.json` + `experiment.patch`.
docs/batched_hillclimb_plan.md	New design/acceptance document for the batched hill-climb system.
.gitignore	Ignores `results/` directory at repo root.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

cgpadwick · 2026-07-08T01:28:47Z

+def bucket_names(storage: Storage, run_id: str) -> set[str]:
+    """Object basenames in the run's mirror prefix (cheap LIST, no GETs)."""
+    try:
+        client = _bucket_client(storage)
+        listed = client.list_objects_v2(Bucket=storage.bucket,
+                                        Prefix=storage.run_prefix(run_id) + "/")
+        return {obj["Key"].rsplit("/", 1)[-1]
+                for obj in listed.get("Contents", [])}
+    except Exception:
+        return set()


Fixed in b573efe — paginated via get_paginator('list_objects_v2'); good catch, a >1000-object mirror would have hidden the marker and stranded the workers.

cgpadwick · 2026-07-08T01:28:48Z

+    stage_root = Path(tempfile.mkdtemp(prefix="saage_batch_"))
+    jobs = []
+    for i, prop in enumerate(proposals):
+        flow_file = stage_job_flow(experiment_flow, prop, workspace,
+                                   stage_root, i)
+        jobs.append(Job(name=f"p{i}", flow_file=str(flow_file),
+                        retry_timeouts=False,   # a too-slow experiment is a
+                                                # result, not a retry
+                        set_args={**(set_args or {}), "job_index": str(i)}))
+
+    extra = {"clock": clock} if clock is not None else {}
+    d = dispatcher_cls(str(experiment_flow), jobs, targets, ops=ops,
+                       provision_cmd=provision_cmd,
+                       provision_files=provision_files,
+                       max_hours=max_hours, poll_interval=poll_interval,
+                       stale_after=stale_after, fetch_dest=results_dir,
+                       dispatch_workers=max(4, len(jobs)),
+                       handoff_opts=opts, **extra)
+    d.run()
+    shutil.rmtree(stage_root, ignore_errors=True)


Fixed in b573efe — staging teardown moved into a finally.

cgpadwick · 2026-07-08T01:28:50Z

+    # the workspace, with the flow dir reachable at ../flow. Serialized under
+    # an flock keyed by the command, so K concurrent bootstraps on one box
+    # populate $SAAGE_CACHE once instead of racing K dataset pulls; with a


Fixed in b573efe — plus the same slip in provision.py.

…up, typos - bucket_names paginates list_objects_v2: a mirror prefix past 1000 objects must not hide batch_done.marker from sweep-watch (workers would never be released) - batch round staging dir is removed in a finally — a dispatch exception no longer leaks the temp tree - 'an flock' -> 'a flock' (scripts.py + the same slip in provision.py) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

# Conflicts: # .gitignore # flows/kaggle_solver/BENCHMARK_RESULTS.md # flows/kaggle_solver/benchmark_journal.jsonl # flows/kaggle_solver/memory/spooky-author-identification.md

cgpadwick and others added 14 commits July 5, 2026 08:53

chore: gitignore results/ (fetched run artifacts, can hold multi-GB c…

8ffc170

…heckpoints) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

docs: README section for the fan-out/fan-in sweep machinery

2bdb0ad

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

cgpadwick requested a review from Copilot July 7, 2026 05:23

Copilot started reviewing on behalf of cgpadwick July 7, 2026 05:24 View session

Copilot AI reviewed Jul 7, 2026

View reviewed changes

cgpadwick and others added 2 commits July 7, 2026 18:28

Merge branch 'feat/awesome-campaign' into feat/parallel-sweep-port

3e48e7c

# Conflicts: # .gitignore # flows/kaggle_solver/BENCHMARK_RESULTS.md # flows/kaggle_solver/benchmark_journal.jsonl # flows/kaggle_solver/memory/spooky-author-identification.md

Navigation Menu

Fan-out/fan-in fleet: parallel-sweep engine recovered from spark, integrated with the MLE-STAR flow, proven live - #37

Fan-out/fan-in fleet: parallel-sweep engine recovered from spark, integrated with the MLE-STAR flow, proven live#37
cgpadwick wants to merge 16 commits into
feat/awesome-campaignfrom
feat/parallel-sweep-port

cgpadwick commented Jul 7, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

cgpadwick Jul 8, 2026

Uh oh!

cgpadwick Jul 8, 2026

Uh oh!

cgpadwick Jul 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Conversation

cgpadwick commented Jul 7, 2026

What this is

The machinery

Proven live (two 5-box sweeps, spooky-author)

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

cgpadwick Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

cgpadwick Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

cgpadwick Jul 8, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants