diff --git a/monarch-benchmark/docs/workflow-builder-prototype-campaign.md b/monarch-benchmark/docs/workflow-builder-prototype-campaign.md
new file mode 100644
index 0000000..ca9f5e1
--- /dev/null
+++ b/monarch-benchmark/docs/workflow-builder-prototype-campaign.md
@@ -0,0 +1,115 @@
+# Workflow builder prototype campaign
+
+The runner is executable without API keys in its default dry-plan mode:
+
+```sh
+uv run python -m wb_orchestrator.prototype_campaign
+```
+
+It reads existing frozen task files, performs local grader checks, and emits a deterministic JSON manifest. It does not create a budget ledger, construct a model competitor, contact a preview, or dispatch work. The manifest includes file hashes, existing contract hashes, grader/runner source hashes, and the AutomationBench revision. Saving stdout produces the selection artifact for review.
+
+## Proposed measurements
+
+Six development fixtures each run against five configurations: current, compiled graph, serial sections, parallel sections with compiled graph, and parallel sections with current graph. Four separate held-out fixtures each run three times against current, serial sections, and parallel sections. This is 30 + 36 = 66 measured attempts, with no automatic retries. All 66 reserve $12 each: $792, alongside the separate $200 implementation/development allowance, leaving $8 unallocated under the $1,000 ceiling. The development *fixtures* are still measured attempts; their costs do not consume the separate engineering allowance.
+
+Development candidates:
+
+- `simple.email_sf_contact_city_update`: single-change baseline.
+- `simple.invoice_airtable_slack`: multiple products.
+- `sales.update_contact_phone`: batch entity matching.
+- `support.freshdesk_auto_merge`: duplicate handling.
+- `support.intercom_freshdesk_escalation`: large policy and cross-product case.
+- `sales.zoom_recording_distribution`: multi-recipient policy and routing.
+
+Held-out candidates:
+
+- `simple.sf_opp_closed_won`: single change with coupled state.
+- `operations.invoice_shipping_trigger`: conditional fulfillment.
+- `support.reamaze_cross_platform_dedup`: cross-product deduplication.
+- `hr.comp_adjustment_batch`: batch eligibility and notifications.
+
+All ten have matching recorded contract hashes, nonempty assertions and expected changes, and reject a no-op. Independent reference actions qualify the two simple Salesforce cases. The invoice case is rejected by demonstrated false positives; seven policy/batch cases remain pending independent references and falsification controls. These are candidate fixtures, not a certified campaign. Assertion counts and task difficulty do not establish authored node counts or section independence.
+
+A separate internal dogfood workflow has been inspected but is not included in this public fixture corpus. These fixtures do not substitute evidence about that workflow. The manifest's `personal_30_node_workflow` missing-fixture marker records that exclusion; the approval record must acknowledge it before an exploratory campaign can proceed.
+
+## Paid preflight
+
+`--execute --proof <file>` refuses missing proof before creating either budget or results databases. It requires a named approval bound to the exact manifest hash, a campaign ID, a PR preview URL, and its full commit. The configured checkout and preview URL must match that approval. No bypass flag exists.
+
+Each evidence reference is `{path, sha256}`. Referenced JSON records must match their checksum and contain `kind`, `verified: true`, and the approved `preview_sha`:
+
+| Proof key | Required kind and evidence fields |
+| --- | --- |
+| `models` | `exact_model_inventory`: `all_roles_accounted_for: true`; `models` maps every enabled role to its exact `provider`, `model_id`, and configured `effort`. |
+| `dollar_enforcement` | `server_dollar_enforcement`: limits `campaign_limit_usd: 1000`, `development_limit_usd: 200`, `attempt_limit_usd: 12`; `inflight_calls_included`, `unknown_usage_blocks`, and `enforced_before_model_calls` all true. |
+| `cancellation` | `settled_cancellation`: `all_child_calls_stopped` and `billing_final` both true. |
+| `world` | `dedicated_synthetic_world`: `per_attempt_reset: true`, `shared_accounts: false`, and the exact configured `front_door_url`. |
+| `graders` | Map every task ID to `task_grader_controls` evidence: exact `task_sha256`, `positive_passed`, `negative_rejected`, and `collateral_rejected` all true. |
+
+These records are reviewed evidence attestations, not something the runner can manufacture from booleans. Their underlying tests and deployment inspection must actually establish the stated properties. The required server dollar enforcement record is currently unavailable. Model and cancellation evidence also need collection. This implementation has not launched a paid campaign.
+
+## Dispatch, accounting, and results
+
+Once verified evidence exists, the runner uses the existing `Orchestrator.from_config`, `MonarchArm`, `Episode` world reset, grader, and result store. It freezes selected task bytes into the result directory for later regrading and saves the approved manifest/proof. Each scheduled task/repetition/configuration entry runs separately, with configuration identity preserved in its run ID and configuration hash. Execution is serial because the existing Monarch front door and fixture routing use a shared fixed port.
+
+An optional orchestrator wrapper checks proof again, reserves the unique attempt ID before each dispatch, and reconciles actual cost afterward. A repeated reservation never dispatches again. Settlement also requires a complete, untruncated durable authoring event export. Every section-model launch must match exactly one later receipt for the same build attempt, phase, section, and retry; every receipt must have a unique call ID, complete usage, and known cost. Build keys can repeat after clarification resumes, so matching follows event order. Ordinary progress and non-model build lifecycle events do not require usage receipts. Missing or partial cost, incomplete child usage, any raised failure/timeout with potentially unsettled work, or cancellation without an explicit successful response becomes unknown liability and stops the campaign. Infrastructure errors do not get automatic paid retries. Recovery requires inspecting outstanding calls and reconciling final usage; restarting a script is not authority to rerun an already reserved attempt.
+
+The SQLite guard is not a provider-side spend limiter. It therefore cannot substitute for the separate server dollar enforcement proof. Use the same `--budget` file as all other paid work in this campaign.
+
+The HTTP front door now releases its fixed fixture port between attempts on
+POSIX systems while still refusing to start over an active listener. The
+socket lifecycle regression passes on macOS and the same lifecycle policy
+addresses the failure observed in Linux CI.
+
+### Independent qualification and execution order
+
+The dry plan now records a seeded paired schedule (`20260907`): each task and
+repetition is a contiguous block containing every compared configuration once.
+Task blocks and the initial configuration order are shuffled; rotations balance
+configuration positions to within one appearance per phase. The existing
+orchestrator runs one attempt per schedule entry, with a unique run ID. This
+keeps the 30 development and 36 holdout comparisons without completing an entire
+configuration before another starts. Warm-cache effects still need measurement;
+order balancing does not make caches identical.
+
+Paid proof must include `resolved_competitor_sha256` in model inventory evidence,
+computed as the campaign's canonical digest of `dataclasses.asdict(competitor)`
+after resolving the product and plan. This binds any local model/effort and
+harness settings; deployed server model roles still require the full inventory.
+The configured Monarch checkout must match the actual deployed revision. For a
+GitHub PR environment that may be the synthetic merge SHA, not branch HEAD:
+check out that exact revision intentionally before collecting evidence.
+
+Run independent, grader-only controls without any model or credentials:
+
+```sh
+uv run python -m wb_orchestrator.prototype_controls --output /tmp/workflow-grader-controls.json
+```
+
+The raw report retains API request bodies, responses, grader results, before/after snapshots, final
+snapshot hashes and task hashes. Runtime-generated timestamps and IDs make this
+report run-specific; the campaign manifest includes stable readiness conclusions
+and hashes the control implementation. Neither the report nor reference answers
+are supplied to workflow authoring. Frozen task contracts and vendor code remain
+unchanged.
+
+Actual keyless findings:
+
+- City update: source email implies Denver; the reference changes Lisa's city.
+  Correct result passes, while also changing her phone fails.
+- Closed Won: the named opportunity changes stage. Correct result passes, while
+  also changing its amount fails. This qualifies the simulated task's requested
+  stage change, not real Salesforce's coupled `is_closed`/`is_won` behavior.
+- Invoice to Airtable and Slack: reject this candidate. The correct CloudHost
+  invoice for $4,500 passes, but so does an invoice with the wrong vendor and
+  amount, and an added record in an unrequested table. Stronger business-result
+  and collateral assertions require the task owner's approval and a new hash.
+- Seven batch/policy cases remain pending, with task-specific reference and
+  falsification work listed in manifest readiness. No suitable replacement has
+  been silently substituted. In particular, the personal 30-node workflow is
+  still missing.
+
+All selected tasks must have local qualified status as well as external approved
+grader evidence before paid dispatch. An attestation cannot override a locally
+rejected or pending candidate. Consequently the currently proposed 66-attempt
+selection remains blocked until qualification and approved reselection are done.
diff --git a/monarch-benchmark/docs/workflow-builder-prototype-client.md b/monarch-benchmark/docs/workflow-builder-prototype-client.md
new file mode 100644
index 0000000..ea69fe6
--- /dev/null
+++ b/monarch-benchmark/docs/workflow-builder-prototype-client.md
@@ -0,0 +1,38 @@
+# Workflow builder prototype testing
+
+Use the existing Monarch competitor with optional `builder_experiment` in its harness YAML: `current`, `compiled`, `sections-serial`, `sections-parallel`, or `sections-parallel-current`. This selection changes the frozen configuration hash. Leaving it unset preserves the existing request body, stream behavior, and hashes.
+
+An explicit selection uses the persisted authoring status endpoint, checks that the server selected the requested experiment, and records raw snapshots and cursor event pages in the attempt log. It waits for `complete: true`, which means the server's terminal state and final event writes are durable. Expired history remains an explicit HTTP 410 failure, including its original response; it never becomes an empty successful export. Existing server payload truncation markers stay visible. Prototype failures retain raw failure fields and are not classified from provider names in prose.
+
+`MonarchClient.cancel_execution(run_id)` uses `/api/engine/runs/:id/cancel` and the existing session header. Prototype execution timeouts call it before cleanup. Authoring cancellation continues using its existing route. A failed cancellation is recorded and requires checking that no paid work remains active.
+
+## Use a hosted Monarch environment
+
+The benchmark can call an existing Monarch deployment; it does not need to run
+its own Monarch server. Remote URLs were already supported by the client. This
+change adds experiment selection and durable evidence to that same API path.
+
+Point `MONARCH_URL` at the selected deployment and configure its benchmark
+session, matching `MONARCH_FD_URL` and catalog credentials, and the existing
+Langfuse cost source. `FRONT_DOOR_URL` must make the benchmark's synthetic HTTP
+world reachable from Monarch. A local loopback address is insufficient when
+Monarch is hosted elsewhere. Keep that world separate from real customer data.
+
+The current harness still reads `monarch_repo` to label the tested version;
+that checkout must match the deployed revision, even though it need not run a
+server. For a PR preview, verify the deployed merge revision rather than
+assuming it equals branch HEAD. Moving an existing lab setup to a hosted
+environment still requires configuring these connections and verifying a
+synthetic attempt; this client change does not perform that migration.
+
+## Reserve the campaign budget before calls
+
+`CampaignBudget(path)` creates a local SQLite ledger with atomic reservations across processes. Its ceiling is $1,000 across the campaign, including at most $200 of development work. An ordinary measured attempt reserves $12. All concurrent in-flight reservations count against the ceilings. Values are rounded upward to integer millionths of a dollar.
+
+1. Use one ledger path for every development call and measured attempt in the campaign.
+2. Call `reserve(stable_attempt_id, 'measured')`, or `reserve(stable_call_id, 'development', amount_usd)` before development work.
+3. Dispatch paid work only when the returned `created` field is true. A repeated reservation ID returns false and never authorizes another dispatch; inspect/recover the prior attempt instead.
+4. Reconcile once with the complete actual provider cost using `reconcile(id, actual_usd)`. Repeating the same final cost is harmless; changing an already reconciled cost is refused. Retries that make additional paid calls need distinct reservation IDs. Each charge belongs to exactly one reservation; do not also reconcile child calls separately when an attempt total includes them.
+5. If cost cannot be established, call `reconcile(id, None)`. Its reserved liability remains charged and every later reservation is blocked until known cost replaces the unknown. Any actual cost exceeding its reservation is recorded honestly and also blocks later work for review.
+
+The helper does not dispatch calls, meter a provider, or cancel a running request. The campaign runner must enforce per-attempt runtime/model limits, reconcile exceptions and crashes, and wait for cancellation to settle before reporting final usage. A reservation alone cannot guarantee that an external call stays within its allocated amount. Existing benchmark grading, task freezes, and report eligibility are unchanged. No campaign is started by these additions.
diff --git a/monarch-benchmark/workflowbench/config/campaigns/workflow-builder-prototype.json b/monarch-benchmark/workflowbench/config/campaigns/workflow-builder-prototype.json
new file mode 100644
index 0000000..32607ae
--- /dev/null
+++ b/monarch-benchmark/workflowbench/config/campaigns/workflow-builder-prototype.json
@@ -0,0 +1,709 @@
+{
+  "attempt_limit_usd": 12,
+  "attempts": 66,
+  "automatic_retries": 0,
+  "automationbench_revision": "4a8e1061254004d9dac807054eed33fad7d1ff14",
+  "campaign_limit_usd": 1000,
+  "development_allowance_usd": 200,
+  "manifest_sha256": "795635158d124f1db68ed98de4c696322d60dec5febc0c16ad443347c0790f97",
+  "maximum_combined_allocation_usd": 992,
+  "missing": [
+    "personal_30_node_workflow"
+  ],
+  "paid_prerequisites": [
+    "approved_manifest",
+    "exact_model_inventory",
+    "server_dollar_enforcement",
+    "settled_cancellation",
+    "dedicated_world",
+    "per_task_grader_evidence"
+  ],
+  "phases": [
+    {
+      "attempts": 30,
+      "configurations": [
+        "current",
+        "compiled",
+        "sections-serial",
+        "sections-parallel",
+        "sections-parallel-current"
+      ],
+      "name": "development",
+      "repetitions": 1,
+      "tasks": 6
+    },
+    {
+      "attempts": 36,
+      "configurations": [
+        "current",
+        "sections-serial",
+        "sections-parallel"
+      ],
+      "name": "holdout",
+      "repetitions": 3,
+      "tasks": 4
+    }
+  ],
+  "qualification": "Candidate fixtures, not evidence of 30-node authoring or representative pilot coverage. Readiness records distinguish qualified, rejected, and pending independent controls; all must qualify before paid dispatch.",
+  "reservation_estimate_usd": 792,
+  "schedule": [
+    {
+      "configuration": "compiled",
+      "phase": "development",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "support.intercom_freshdesk_escalation"
+    },
+    {
+      "configuration": "current",
+      "phase": "development",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "support.intercom_freshdesk_escalation"
+    },
+    {
+      "configuration": "sections-parallel-current",
+      "phase": "development",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "support.intercom_freshdesk_escalation"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "development",
+      "position": 3,
+      "repetition": 1,
+      "task_id": "support.intercom_freshdesk_escalation"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "development",
+      "position": 4,
+      "repetition": 1,
+      "task_id": "support.intercom_freshdesk_escalation"
+    },
+    {
+      "configuration": "current",
+      "phase": "development",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "sales.update_contact_phone"
+    },
+    {
+      "configuration": "sections-parallel-current",
+      "phase": "development",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "sales.update_contact_phone"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "development",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "sales.update_contact_phone"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "development",
+      "position": 3,
+      "repetition": 1,
+      "task_id": "sales.update_contact_phone"
+    },
+    {
+      "configuration": "compiled",
+      "phase": "development",
+      "position": 4,
+      "repetition": 1,
+      "task_id": "sales.update_contact_phone"
+    },
+    {
+      "configuration": "sections-parallel-current",
+      "phase": "development",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "simple.email_sf_contact_city_update"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "development",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "simple.email_sf_contact_city_update"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "development",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "simple.email_sf_contact_city_update"
+    },
+    {
+      "configuration": "compiled",
+      "phase": "development",
+      "position": 3,
+      "repetition": 1,
+      "task_id": "simple.email_sf_contact_city_update"
+    },
+    {
+      "configuration": "current",
+      "phase": "development",
+      "position": 4,
+      "repetition": 1,
+      "task_id": "simple.email_sf_contact_city_update"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "development",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "support.freshdesk_auto_merge"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "development",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "support.freshdesk_auto_merge"
+    },
+    {
+      "configuration": "compiled",
+      "phase": "development",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "support.freshdesk_auto_merge"
+    },
+    {
+      "configuration": "current",
+      "phase": "development",
+      "position": 3,
+      "repetition": 1,
+      "task_id": "support.freshdesk_auto_merge"
+    },
+    {
+      "configuration": "sections-parallel-current",
+      "phase": "development",
+      "position": 4,
+      "repetition": 1,
+      "task_id": "support.freshdesk_auto_merge"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "development",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "sales.zoom_recording_distribution"
+    },
+    {
+      "configuration": "compiled",
+      "phase": "development",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "sales.zoom_recording_distribution"
+    },
+    {
+      "configuration": "current",
+      "phase": "development",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "sales.zoom_recording_distribution"
+    },
+    {
+      "configuration": "sections-parallel-current",
+      "phase": "development",
+      "position": 3,
+      "repetition": 1,
+      "task_id": "sales.zoom_recording_distribution"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "development",
+      "position": 4,
+      "repetition": 1,
+      "task_id": "sales.zoom_recording_distribution"
+    },
+    {
+      "configuration": "compiled",
+      "phase": "development",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "simple.invoice_airtable_slack"
+    },
+    {
+      "configuration": "current",
+      "phase": "development",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "simple.invoice_airtable_slack"
+    },
+    {
+      "configuration": "sections-parallel-current",
+      "phase": "development",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "simple.invoice_airtable_slack"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "development",
+      "position": 3,
+      "repetition": 1,
+      "task_id": "simple.invoice_airtable_slack"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "development",
+      "position": 4,
+      "repetition": 1,
+      "task_id": "simple.invoice_airtable_slack"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 3,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 3,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 3,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 2,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 2,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 2,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 2,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 2,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 2,
+      "task_id": "hr.comp_adjustment_batch"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 2,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 2,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 2,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 2,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 2,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 2,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 3,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 3,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 3,
+      "task_id": "support.reamaze_cross_platform_dedup"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 3,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 3,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 3,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 3,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 3,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 3,
+      "task_id": "simple.sf_opp_closed_won"
+    },
+    {
+      "configuration": "current",
+      "phase": "holdout",
+      "position": 0,
+      "repetition": 1,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "sections-serial",
+      "phase": "holdout",
+      "position": 1,
+      "repetition": 1,
+      "task_id": "operations.invoice_shipping_trigger"
+    },
+    {
+      "configuration": "sections-parallel",
+      "phase": "holdout",
+      "position": 2,
+      "repetition": 1,
+      "task_id": "operations.invoice_shipping_trigger"
+    }
+  ],
+  "schedule_seed": 20260907,
+  "source_sha256": {
+    "grader/grade.py": "86214ac4ff260b06e3a9ca9e9dfc44eac141dcf3ae2ad3b05c9ca8db381488b1",
+    "grader/invariant.py": "507e0d34f8ea7a4c7ef1191d307e7abeb7b58efeee18547cb1b00966b45a4cd4",
+    "runner/arms.py": "36a68fe243a74b0b52faf26fc5434a889969485160b91b58777d71ef0962a86b",
+    "wb_arms/monarch.py": "b1b815da33800b942d06156d64df84de9a98f505ff72fbc2160546e80310c3af",
+    "wb_arms/monarch_client.py": "9ac1d23aa8f8884ae24a9ad62f2c424e263bbe7dff8f9862da1a242795b819b8",
+    "wb_orchestrator/campaign_budget.py": "d05926af033f87494c3081695c17a5401264f4cf82c16dc310243a23301c1c5c",
+    "wb_orchestrator/config.py": "605b5fc31ff7bbbdbadbd8482b6425469b745a0f2845da1b148ed89cbdeb498c",
+    "wb_orchestrator/orchestrator.py": "bd1657ecd0785f62478861d5f9a7a1a105f0b0acf175f71f84491c3812a7474a",
+    "wb_orchestrator/prototype_campaign.py": "eaa312bd1a845ae6ebe9548c3be5b6b901b0ba97dd093639d25e63e6c45ddf9f",
+    "wb_orchestrator/prototype_controls.py": "2e7018cc2b652b0936d494b2ae11304230e672f3cdffb5fd0d9088a84ec82a53",
+    "wb_world/episode.py": "60f794d4a2634fe3da07ff7da0d933fb25174ca27e3cab1cf5dd4c9165350c72"
+  },
+  "tasks": [
+    {
+      "contract_hash": "d44c785ac30f7594",
+      "file_sha256": "4e1dfa898186891cd79efcf5aa6a382831c45bb48b28b4abc33a44ee0241ea55",
+      "id": "simple.email_sf_contact_city_update",
+      "path": "tasks/simple.email_sf_contact_city_update.json",
+      "purpose": "single change baseline",
+      "readiness": {
+        "assertions": 1,
+        "collateral_control": "rejected",
+        "expected_changes": 1,
+        "negative_control": "rejected",
+        "positive_control": "passed",
+        "source_contract_matches": true,
+        "status": "qualified"
+      },
+      "split": "development"
+    },
+    {
+      "contract_hash": "eeb2e22aa27c8fcd",
+      "file_sha256": "acbdb2ff67966b99da7b56a57227fde1e4fc814570b2a010fc60563fee5cdc8d",
+      "id": "simple.invoice_airtable_slack",
+      "path": "tasks/tier-medium/simple.invoice_airtable_slack.json",
+      "purpose": "multiple products",
+      "readiness": {
+        "assertions": 2,
+        "collateral_control": "accepted",
+        "expected_changes": 2,
+        "negative_control": "rejected",
+        "positive_control": "passed",
+        "remaining": "Frozen grader accepts wrong vendor and amount; review and approve stronger assertions before selecting this task.",
+        "semantic_counterexample": "accepted",
+        "source_contract_matches": true,
+        "status": "rejected"
+      },
+      "split": "development"
+    },
+    {
+      "contract_hash": "39a052c6cea301c4",
+      "file_sha256": "8c4a40803b43f3aa8c4d483916c5fd3bbb4e55fc782033c3d37aaadc67637bb6",
+      "id": "sales.update_contact_phone",
+      "path": "tasks/tier-simple/sales.update_contact_phone.json",
+      "purpose": "batch entity matching",
+      "readiness": {
+        "assertions": 8,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 6,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Implement HR batch verification, cancellation, duplicate-contact matching, and note reference; falsify unrelated contact-field edits permitted by broad expected_changes.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "development"
+    },
+    {
+      "contract_hash": "12218a5f54fcdde4",
+      "file_sha256": "79c675d471afa71fb1ceefeeb23e6864c7f15fe689c2e8f2d2afc0c6f3078c37",
+      "id": "support.freshdesk_auto_merge",
+      "path": "tasks/tier-medium/support.freshdesk_auto_merge.json",
+      "purpose": "duplicate handling",
+      "readiness": {
+        "assertions": 26,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 20,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Independently implement ss_merge_rules/ws_rules with same-requester restriction; falsify cross-requester merges and unrelated ticket changes.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "development"
+    },
+    {
+      "contract_hash": "6b64e12460e6d999",
+      "file_sha256": "abac0a4fa60424d8dcc0949ed07d492e1950da57defad508ff2c8dda48d64ba4",
+      "id": "support.intercom_freshdesk_escalation",
+      "path": "tasks/tier-complex/support.intercom_freshdesk_escalation.json",
+      "purpose": "large policy and cross-product case",
+      "readiness": {
+        "assertions": 56,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 18,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Independently apply ss_escalation_config, then check destinations, confirmation replies, exclusions, and untouched conversations.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "development"
+    },
+    {
+      "contract_hash": "ca1b00a0d5694f9f",
+      "file_sha256": "978d6758f32f7c8fe30b5c4418c10b43450bf179d99463065a774ddb45d71e36",
+      "id": "sales.zoom_recording_distribution",
+      "path": "tasks/tier-complex/sales.zoom_recording_distribution.json",
+      "purpose": "multi-recipient policy and routing",
+      "readiness": {
+        "assertions": 16,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 3,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Independently resolve yesterday completed meetings and distribution policy; falsify wrong audiences, excluded meetings, and duplicate sends.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "development"
+    },
+    {
+      "contract_hash": "e4a3adbe074495d5",
+      "file_sha256": "1d9643621fbeac83685946d8a846936c10c0829f724275d8a20cc2684c33a6c1",
+      "id": "simple.sf_opp_closed_won",
+      "path": "tasks/simple.sf_opp_closed_won.json",
+      "purpose": "single change with coupled state",
+      "readiness": {
+        "assertions": 1,
+        "collateral_control": "rejected",
+        "expected_changes": 1,
+        "negative_control": "rejected",
+        "positive_control": "passed",
+        "source_contract_matches": true,
+        "status": "qualified"
+      },
+      "split": "holdout"
+    },
+    {
+      "contract_hash": "afcbaa6c977eb3b2",
+      "file_sha256": "c2f878ff5ae6b45f03dcab9258acd5bc69c90b86f477b6ed852dec1a9a7fee6f",
+      "id": "operations.invoice_shipping_trigger",
+      "path": "tasks/tier-medium/operations.invoice_shipping_trigger.json",
+      "purpose": "conditional fulfillment",
+      "readiness": {
+        "assertions": 9,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 4,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Independently calculate only -EXP line totals and counts; verify Monday fields and warehouse email, then falsify non-expedited inclusion.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "holdout"
+    },
+    {
+      "contract_hash": "f9b1fffe98a7d32c",
+      "file_sha256": "4ff90629d1b5cdcd704161492b1f6a6778e7ad9f96e01e864d534da21a2396e6",
+      "id": "support.reamaze_cross_platform_dedup",
+      "path": "tasks/random-10/support.reamaze_cross_platform_dedup.json",
+      "purpose": "cross-product deduplication",
+      "readiness": {
+        "assertions": 36,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 19,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Independently join customer aliases and issue identity; verify reciprocal notes, close state and log, then falsify unrelated-customer closure.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "holdout"
+    },
+    {
+      "contract_hash": "9ebf85d40fdc6e31",
+      "file_sha256": "db35b2b7b026bf321460d8f6781ee6ea8fa625b626fd4acb651f1458cf813a0b",
+      "id": "hr.comp_adjustment_batch",
+      "path": "tasks/tier-medium/hr.comp_adjustment_batch.json",
+      "purpose": "batch eligibility and notifications",
+      "readiness": {
+        "assertions": 21,
+        "collateral_control": "pending_independent_reference",
+        "expected_changes": 3,
+        "negative_control": "rejected",
+        "positive_control": "pending_independent_reference",
+        "remaining": "Independently resolve cleared rows and current procedures; verify computed raises and employee/manager notifications, then falsify ineligible-row changes.",
+        "source_contract_matches": true,
+        "status": "pending"
+      },
+      "split": "holdout"
+    }
+  ],
+  "version": 1
+}
diff --git a/monarch-benchmark/workflowbench/tests/monarch_helpers.py b/monarch-benchmark/workflowbench/tests/monarch_helpers.py
index a8a2db2..8347c6d 100644
--- a/monarch-benchmark/workflowbench/tests/monarch_helpers.py
+++ b/monarch-benchmark/workflowbench/tests/monarch_helpers.py
@@ -5,6 +5,7 @@
 """
 from __future__ import annotations
 
+import os
 import socket
 import subprocess
 
@@ -66,6 +67,8 @@ def free_port() -> int:
 def free(port: int) -> None:
     """Assert the front door let go of its fixed port."""
     s = socket.socket()
+    if os.name != "nt" and hasattr(socket, "SO_REUSEADDR"):
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     s.bind(("0.0.0.0", port))
     s.close()
 
diff --git a/monarch-benchmark/workflowbench/tests/test_campaign_budget.py b/monarch-benchmark/workflowbench/tests/test_campaign_budget.py
new file mode 100644
index 0000000..9d9983f
--- /dev/null
+++ b/monarch-benchmark/workflowbench/tests/test_campaign_budget.py
@@ -0,0 +1,89 @@
+from concurrent.futures import ThreadPoolExecutor
+from decimal import Decimal
+
+import pytest
+
+from wb_orchestrator.campaign_budget import CampaignBudget, BudgetBlocked, ReservationConflict
+
+
+def test_atomic_reservations_cannot_oversubscribe(tmp_path):
+    path = tmp_path / 'budget.sqlite'
+    CampaignBudget(path, total_limit_usd=24)
+    def reserve(i):
+        try:
+            return CampaignBudget(path, total_limit_usd=24).reserve(str(i), 'measured').created
+        except BudgetBlocked:
+            return False
+    with ThreadPoolExecutor(max_workers=8) as pool:
+        assert sum(pool.map(reserve, range(20))) == 2
+    assert CampaignBudget(path, total_limit_usd=24).snapshot()['reserved_usd'] == 24
+
+
+def test_retries_and_reconciliation_do_not_double_count(tmp_path):
+    b = CampaignBudget(tmp_path / 'b.sqlite')
+    assert b.reserve('attempt', 'measured').created
+    assert not b.reserve('attempt', 'measured').created
+    b.reconcile('attempt', Decimal('1.23'))
+    b.reconcile('attempt', Decimal('1.23'))
+    assert b.snapshot()['spent_usd'] == 1.23
+    assert b.snapshot()['reserved_usd'] == 0
+    assert not b.reserve('attempt', 'measured').created
+    with pytest.raises(ReservationConflict):
+        b.reconcile('attempt', 2)
+
+
+def test_unknown_usage_blocks_new_paid_work_until_resolved(tmp_path):
+    b = CampaignBudget(tmp_path / 'b.sqlite')
+    b.reserve('attempt', 'measured')
+    b.reconcile('attempt', None)
+    with pytest.raises(BudgetBlocked):
+        b.reserve('next', 'measured')
+    assert b.snapshot()['unknown'] == ['attempt']
+    assert b.snapshot()['reserved_usd'] == 12
+    b.reconcile('attempt', 5)
+    assert b.reserve('next', 'measured').created
+
+
+def test_development_and_campaign_ceiling_include_pending_work(tmp_path):
+    b = CampaignBudget(tmp_path / 'b.sqlite')
+    b.reserve('dev1', 'development', 200)
+    with pytest.raises(BudgetBlocked):
+        b.reserve('dev2', 'development', 1)
+    b.reconcile('dev1', 200)
+    for i in range(66):
+        b.reserve(f'm{i}', 'measured')
+    with pytest.raises(BudgetBlocked):
+        b.reserve('overflow', 'measured')
+    assert b.snapshot()['committed_usd'] == 992
+
+
+def test_overrun_is_recorded_and_stops_later_calls(tmp_path):
+    b = CampaignBudget(tmp_path / 'b.sqlite')
+    b.reserve('attempt', 'measured')
+    b.reconcile('attempt', 13)
+    assert b.snapshot()['spent_usd'] == 13
+    with pytest.raises(BudgetBlocked):
+        b.reserve('next', 'measured')
+
+
+def test_conflicting_reservations_and_budget_configuration_refused(tmp_path):
+    path = tmp_path / 'b.sqlite'
+    b = CampaignBudget(path)
+    b.reserve('same', 'development', 5)
+    with pytest.raises(ReservationConflict):
+        b.reserve('same', 'measured')
+    with pytest.raises(ReservationConflict):
+        CampaignBudget(path, total_limit_usd=500)
+    with pytest.raises(ValueError):
+        CampaignBudget(tmp_path / 'other.sqlite', total_limit_usd=1001)
+    with pytest.raises(ValueError):
+        b.reserve('measured', 'measured', 13)
+
+
+@pytest.mark.parametrize('amount', [-1, float('nan'), float('inf'), True])
+def test_invalid_cost_cannot_change_the_ledger(tmp_path, amount):
+    b = CampaignBudget(tmp_path / 'b.sqlite')
+    b.reserve('attempt', 'measured')
+    with pytest.raises(ValueError):
+        b.reconcile('attempt', amount)
+    assert b.snapshot()['reserved_usd'] == 12
diff --git a/monarch-benchmark/workflowbench/tests/test_config.py b/monarch-benchmark/workflowbench/tests/test_config.py
index 815e638..8fa4821 100644
--- a/monarch-benchmark/workflowbench/tests/test_config.py
+++ b/monarch-benchmark/workflowbench/tests/test_config.py
@@ -879,3 +879,12 @@ def write(mapping):
     with pytest.raises(ConfigError) as exc:
         config.load_monarch_kb(write(underscored), product)
     assert "no entry" in str(exc.value) and "-" in exc.value.field
+
+
+def test_builder_experiment_is_allowlisted_and_only_explicit_selection_changes_hash(tmp_path):
+    original = config.load_harness(write(tmp_path, HARNESS_MONARCH))
+    assert original.builder_experiment is None
+    assert 'builder_experiment' not in config._hashed_harness(original)
+    selected = config.load_harness(write(tmp_path, HARNESS_MONARCH + '\nbuilder_experiment: sections-parallel\n'))
+    assert config._hashed_harness(selected)['builder_experiment'] == 'sections-parallel'
+    check_error(config.load_harness, write(tmp_path, HARNESS_MONARCH + '\nbuilder_experiment: typo\n'), 'builder_experiment')
diff --git a/monarch-benchmark/workflowbench/tests/test_edge_cases.py b/monarch-benchmark/workflowbench/tests/test_edge_cases.py
index ba21445..126d91b 100644
--- a/monarch-benchmark/workflowbench/tests/test_edge_cases.py
+++ b/monarch-benchmark/workflowbench/tests/test_edge_cases.py
@@ -13,6 +13,7 @@
 from tests.test_m1 import make_orch
 from wb_orchestrator.orchestrator import Orchestrator, RunKilled, regrade
 from wb_results.store import Store
+from wb_arms.api_loop import ArmResult, InfraError
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -60,6 +61,24 @@ def test_spend_survives_exhausted_infra(tmp_path, mock_server):
     assert r["cost_usd"] > 0
 
 
+def test_zero_cost_infra_partial_preserves_failure_evidence_in_store(tmp_path, mock_server):
+    store, orch = make_orch(tmp_path, k=1, concurrency=1)
+    orch.tasks = orch.tasks[:1]
+    partial = ArmResult(turn_log=[{'monarch': {'recipeRunId': 'recipe-run'}},
+                                  {'authoring_cancel_error': {'ok': False}}],
+                        flags=['cost_missing'])
+    error = InfraError('infra:harness_crash', 'poll HTTP 500', retryable=False)
+    error.partial = partial
+    arm = type('Arm', (), {'name': 'prototype', 'provider_key': None,
+                           'run': lambda self, ep, deadline: (_ for _ in ()).throw(error)})()
+    orch.arm_wrapper = lambda _: arm
+    run_id = orch.run('run-zero-cost-evidence')
+    row = store.episodes(run=run_id)['rows'][0]
+    assert row['flags'] == ['cost_missing']
+    turns = Path(store.artifacts(row['episode_id'])['turns']).read_text()
+    assert 'recipeRunId' in turns and 'authoring_cancel_error' in turns
+
+
 # -- protocol malformations don't kill episodes -------------------------------
 
 def test_malformed_tool_args_fed_back_to_model(tmp_path, mock_server):
diff --git a/monarch-benchmark/workflowbench/tests/test_monarch_arm.py b/monarch-benchmark/workflowbench/tests/test_monarch_arm.py
index dbefd49..37f8680 100644
--- a/monarch-benchmark/workflowbench/tests/test_monarch_arm.py
+++ b/monarch-benchmark/workflowbench/tests/test_monarch_arm.py
@@ -12,7 +12,6 @@
 from __future__ import annotations
 
 import json
-import socket
 import subprocess
 import threading
 import time
@@ -122,9 +121,7 @@ def test_completed_attempt(site, repo):
     assert result.phases["authoring"].wall_clock_s > 0
     assert result.phases["execution"].wall_clock_s > 0
     # The front door let go of its fixed port.
-    s = socket.socket()
-    s.bind(("0.0.0.0", port))
-    s.close()
+    free(port)
 
 
 # -- T027/T028: one Monarch at a time, and a port that is already taken -------
@@ -353,9 +350,7 @@ def test_termination_table(site, repo, kwargs, expected):
     # Only a `done` frame that actually names a workflow leaves one to delete.
     authored = any(r["path"] == "/api/workflows/recipe/runs" for r in fake.requests)         and any(f.get("status") == "done" and f.get("workflowId") for f in sc.frames)
     assert fake.deleted_workflows == (["wf-1"] if authored else [])
-    s = socket.socket()
-    s.bind(("0.0.0.0", port))
-    s.close()
+    free(port)
 
 
 # -- T035/T036: the deadline, in either phase ---------------------------------
diff --git a/monarch-benchmark/workflowbench/tests/test_monarch_prototype_client.py b/monarch-benchmark/workflowbench/tests/test_monarch_prototype_client.py
new file mode 100644
index 0000000..da67c48
--- /dev/null
+++ b/monarch-benchmark/workflowbench/tests/test_monarch_prototype_client.py
@@ -0,0 +1,176 @@
+import time
+from unittest.mock import Mock
+
+import pytest
+
+from wb_arms.api_loop import EpisodeTimeout, InfraError
+from wb_arms.monarch_client import MonarchClient, MonarchRefused
+
+
+def client():
+    c = MonarchClient('http://test', 'session')
+    c._call = Mock()
+    return c
+
+
+def test_optional_experiment_is_sent_without_changing_default_body():
+    c = client()
+    c._call.return_value = {'runId': 'r'}
+    c.start_authoring('request', 'episode', experiment='sections-parallel')
+    assert c._call.call_args.args[2] == {'goal': 'request', 'experiment': 'sections-parallel'}
+    c.start_authoring('request', 'episode')
+    assert c._call.call_args.args[2] == {'goal': 'request'}
+
+
+def test_authoring_poll_keeps_raw_failure_evidence():
+    c = client()
+    raw = {'status': 'error', 'failure': {'code': None, 'message': 'Provider mystery'}}
+    c._call.return_value = raw
+    assert c.get_authoring('r') is raw
+    assert c._call.call_args.args[:2] == ('GET', '/api/workflows/recipe/runs/r')
+
+
+def test_export_waits_for_durable_terminal_and_retains_every_raw_page():
+    c = client()
+    pages = [
+        {'runId': 'r', 'events': [{'seq': 1, 'data': {'truncated': True}}], 'nextAfterSeq': 1, 'hasMore': False, 'complete': False},
+        {'runId': 'r', 'events': [{'seq': 2}], 'nextAfterSeq': 2, 'hasMore': True, 'complete': False},
+        {'runId': 'r', 'events': [{'seq': 3}], 'nextAfterSeq': 3, 'hasMore': False, 'complete': True},
+    ]
+    c._call.side_effect = pages
+    assert list(c.authoring_event_pages('r', deadline=time.monotonic() + 10, poll_interval_s=0)) == pages
+    assert 'afterSeq=1' in c._call.call_args_list[1].args[1]
+
+
+def test_expired_events_are_not_silently_converted_to_empty_history():
+    c = client()
+    c._call.side_effect = MonarchRefused('RECIPE_RUN_EVENTS_EXPIRED', 410, {'code': 'RECIPE_RUN_EVENTS_EXPIRED'})
+    with pytest.raises(MonarchRefused) as exc:
+        list(c.authoring_event_pages('r', deadline=time.monotonic() + 10))
+    assert exc.value.status == 410
+
+
+def test_invalid_event_page_cannot_claim_complete_trace():
+    c = client()
+    c._call.return_value = {'runId': 'r', 'events': [], 'nextAfterSeq': 0, 'hasMore': True, 'complete': True}
+    with pytest.raises(InfraError):
+        list(c.authoring_event_pages('r', deadline=time.monotonic() + 10))
+
+
+def test_expired_deadline_makes_no_read():
+    c = client()
+    with pytest.raises(EpisodeTimeout):
+        list(c.authoring_event_pages('r', deadline=time.monotonic() - 1))
+    c._call.assert_not_called()
+
+
+def test_execution_cancel_uses_existing_session_route():
+    c = client()
+    c.cancel_execution('engine-run')
+    assert c._call.call_args.args[:3] == ('POST', '/api/engine/runs/engine-run/cancel', {})
+
+
+def test_prototype_competitor_polls_and_keeps_raw_events():
+    from types import SimpleNamespace
+    from wb_arms.monarch import MonarchArm
+    from wb_arms.api_loop import ArmResult
+    c = client()
+    c.start_authoring = Mock(return_value='r')
+    frame = {'status': 'done', 'workflowId': 'wf', 'recipeVersion': 1, 'experiment': {'id': 'compiled'}}
+    c.authoring_snapshots = Mock(return_value=iter([frame]))
+    page = {'runId': 'r', 'events': [], 'nextAfterSeq': 0, 'hasMore': False, 'complete': True}
+    c.authoring_event_pages = Mock(return_value=iter([page]))
+    h = SimpleNamespace(authoring_mode='interactive', builder_experiment='compiled')
+    competitor = MonarchArm(h, 10, None, None, {}, 'prototype')
+    result, ids = ArmResult(), {}
+    assert competitor._author(c, None, 'request', time.monotonic() + 10, result, ids) == 'wf'
+    assert c.start_authoring.call_args.kwargs['experiment'] == 'compiled'
+    assert {'frame': frame} in result.turn_log
+    assert {'authoring_events': page} in result.turn_log
+
+
+def test_prototype_competitor_refuses_a_silently_changed_configuration():
+    from types import SimpleNamespace
+    from wb_arms.monarch import MonarchArm
+    from wb_arms.api_loop import ArmResult
+    c = client()
+    c.start_authoring = Mock(return_value='r')
+    c.authoring_snapshots = Mock(return_value=iter([{'status': 'done', 'experiment': {'id': 'current'}}]))
+    h = SimpleNamespace(authoring_mode='interactive', builder_experiment='compiled')
+    competitor = MonarchArm(h, 10, None, None, {}, 'prototype')
+    with pytest.raises(InfraError, match='Persisted builder experiment'):
+        competitor._author(c, None, 'request', time.monotonic() + 10, ArmResult(), {})
+
+
+def test_unknown_authoring_error_is_not_classified_by_provider_words():
+    from types import SimpleNamespace
+    from wb_arms.monarch import MonarchArm
+    from wb_arms.api_loop import ArmResult
+    c = client()
+    c.start_authoring = Mock(return_value='r')
+    frame = {'status': 'error', 'error': 'AWS credential unknown', 'failure': {'code': None}, 'experiment': {'id': 'compiled'}}
+    c.authoring_snapshots = Mock(return_value=iter([frame]))
+    c.authoring_event_pages = Mock(return_value=iter([]))
+    h = SimpleNamespace(authoring_mode='interactive', builder_experiment='compiled')
+    competitor = MonarchArm(h, 10, None, None, {}, 'prototype')
+    result = ArmResult()
+    assert competitor._author(c, None, 'request', time.monotonic() + 10, result, {}) is None
+    assert competitor._infra is None
+    assert {'frame': frame} in result.turn_log
+
+
+def test_prototype_authoring_transport_failure_attempts_cancel_and_records_failure():
+    from types import SimpleNamespace
+    from wb_arms.monarch import MonarchArm
+    from wb_arms.api_loop import ArmResult
+    c = client()
+    c.start_authoring = Mock(return_value='r')
+    c.authoring_snapshots = Mock(side_effect=InfraError('infra:harness_crash', 'HTTP 500'))
+    c.cancel = Mock(side_effect=InfraError('infra:harness_crash', 'cancel HTTP 500'))
+    h = SimpleNamespace(authoring_mode='interactive', builder_experiment='compiled')
+    competitor = MonarchArm(h, 10, None, None, {}, 'prototype')
+    result = ArmResult()
+    with pytest.raises(InfraError, match='HTTP 500'):
+        competitor._author(c, None, 'request', time.monotonic() + 10, result, {})
+    c.cancel.assert_called_once()
+    assert any('authoring_cancel_error' in row for row in result.turn_log)
+
+
+@pytest.mark.parametrize('response, confirmed', [
+    ({'ok': True, 'status': 'cancelled'}, True),
+    ({'ok': False, 'status': 'running'}, False),
+    ({'status': 'cancelled'}, False),
+])
+def test_prototype_authoring_cancel_requires_explicit_confirmation(response, confirmed):
+    from types import SimpleNamespace
+    from wb_arms.monarch import MonarchArm
+    from wb_arms.api_loop import ArmResult
+    c = client()
+    c.cancel = Mock(return_value=response)
+    competitor = MonarchArm(SimpleNamespace(builder_experiment='compiled'), 10,
+                            None, None, {}, 'prototype')
+    result = ArmResult()
+    competitor._cancel_authoring_prototype(c, 'r', result)
+    assert ('authoring_cancel' in result.turn_log[0]) is confirmed
+    assert ('authoring_cancel_error' in result.turn_log[0]) is not confirmed
+    assert response in result.turn_log[0].values()
+
+
+def test_prototype_execution_transport_failure_cancels_known_run_and_records_failure():
+    from types import SimpleNamespace
+    from wb_arms.monarch import MonarchArm
+    from wb_arms.api_loop import ArmResult
+    c = client()
+    c.get_workflow = Mock(return_value={'recipe': {}})
+    c.run_workflow = Mock(return_value={'id': 'engine-run'})
+    c.get_run = Mock(side_effect=InfraError('infra:harness_crash', 'poll HTTP 500'))
+    c.cancel_execution = Mock(side_effect=InfraError('infra:harness_crash', 'cancel HTTP 500'))
+    h = SimpleNamespace(builder_experiment='compiled')
+    competitor = MonarchArm(h, 10, None, None, {}, 'prototype')
+    result, ids = ArmResult(), {'recipeVersion': None}
+    with pytest.raises(InfraError, match='poll HTTP 500'):
+        competitor._execute(c, SimpleNamespace(task={}), 'wf', time.monotonic() + 10,
+                            result, ids)
+    c.cancel_execution.assert_called_once()
+    assert ids['runId'] == 'engine-run'
+    assert any('execution_cancel_error' in row for row in result.turn_log)
diff --git a/monarch-benchmark/workflowbench/tests/test_openapi_shim.py b/monarch-benchmark/workflowbench/tests/test_openapi_shim.py
index 79fd96d..4137b8d 100644
--- a/monarch-benchmark/workflowbench/tests/test_openapi_shim.py
+++ b/monarch-benchmark/workflowbench/tests/test_openapi_shim.py
@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import json
+import socket
 import urllib.request
 from pathlib import Path
 
@@ -95,3 +96,31 @@ def test_default_host_stays_loopback():
         assert s.url == f"http://127.0.0.1:{s.port}"
     finally:
         s.httpd.server_close()
+
+
+def test_fixed_port_can_restart_after_serving_a_request():
+    ep = Episode(load_task_file(TASK), episode_id="shim-restart")
+    first = EpisodeHTTPShim(ep).start()
+    port = first.port
+    status, _ = _http("GET", f"{first.url}/openapi/index.json")
+    assert status == 200
+    first.stop()
+
+    second = EpisodeHTTPShim(ep, port=port).start()
+    try:
+        status, _ = _http("GET", f"{second.url}/openapi/index.json")
+        assert status == 200
+    finally:
+        second.stop()
+
+
+def test_fixed_port_refuses_an_active_listener():
+    listener = socket.socket()
+    listener.bind(("127.0.0.1", 0))
+    listener.listen()
+    try:
+        ep = Episode(load_task_file(TASK), episode_id="shim-busy")
+        with pytest.raises(OSError):
+            EpisodeHTTPShim(ep, port=listener.getsockname()[1])
+    finally:
+        listener.close()
diff --git a/monarch-benchmark/workflowbench/tests/test_prototype_campaign.py b/monarch-benchmark/workflowbench/tests/test_prototype_campaign.py
new file mode 100644
index 0000000..7c4645e
--- /dev/null
+++ b/monarch-benchmark/workflowbench/tests/test_prototype_campaign.py
@@ -0,0 +1,342 @@
+import json
+import time
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from wb_arms.api_loop import ArmResult, EpisodeTimeout, InfraError
+from wb_orchestrator.campaign_budget import CampaignBudget
+from wb_orchestrator.prototype_campaign import (BudgetedCompetitor, PreflightError, build_manifest,
+                                               durable_section_usage_complete, main,
+                                               validate_preflight, ROOT)
+
+
+def test_manifest_is_keyless_deterministic_and_has_exact_approved_attempt_counts():
+    first, second = build_manifest(), build_manifest()
+    assert first == second
+    assert first['attempts'] == 66
+    assert [p['attempts'] for p in first['phases']] == [30, 36]
+    assert first['reservation_estimate_usd'] == 792
+    assert first['missing'] == ['personal_30_node_workflow']
+    assert len({t['id'] for t in first['tasks']}) == 10
+    assert all(len(t['file_sha256']) == 64 for t in first['tasks'])
+    assert all(t['readiness']['negative_control'] == 'rejected' for t in first['tasks'])
+
+
+def test_dry_plan_does_not_create_budget_or_construct_a_model_competitor(tmp_path, monkeypatch, capsys):
+    monkeypatch.setattr('wb_orchestrator.prototype_campaign.CampaignBudget', Mock(side_effect=AssertionError('ledger created')))
+    assert main(['--budget', str(tmp_path/'budget.sqlite')]) == 0
+    assert json.loads(capsys.readouterr().out)['attempts'] == 66
+    assert not (tmp_path/'budget.sqlite').exists()
+
+
+def test_missing_proof_prevents_execution_before_any_ledger_mutation(tmp_path):
+    with pytest.raises(PreflightError):
+        main(['--execute', '--budget', str(tmp_path/'budget.sqlite')])
+    assert not (tmp_path/'budget.sqlite').exists()
+
+
+def test_preflight_cannot_use_a_manifest_with_different_hashes(tmp_path):
+    p = tmp_path/'proof.json'
+    p.write_text(json.dumps({'manifest_sha256': 'wrong'}))
+    with pytest.raises(PreflightError, match='manifest'):
+        validate_preflight(p, build_manifest())
+
+
+def competitor(tmp_path, result=None):
+    budget = CampaignBudget(tmp_path/'budget.sqlite')
+    complete = ArmResult(cost_usd=2, turn_log=[
+        {'authoring_events': {'complete': True, 'events': []}},
+    ])
+    inner = SimpleNamespace(name='model', provider_key='monarch', run=Mock(return_value=result or complete))
+    stop = Mock()
+    return BudgetedCompetitor(inner, budget, authorize=lambda: None, stop=stop), inner, budget, stop
+
+
+def test_reservation_precedes_dispatch_and_duplicate_id_never_dispatches_twice(tmp_path):
+    wrapped, inner, budget, _ = competitor(tmp_path)
+    def run(ep, deadline):
+        assert budget.snapshot()['reserved_usd'] == 12
+        return ArmResult(cost_usd=2, turn_log=[
+            {'authoring_events': {'complete': True, 'events': []}},
+        ])
+    inner.run.side_effect = run
+    ep = SimpleNamespace(episode_id='stable-attempt')
+    wrapped.run(ep, time.monotonic()+1)
+    assert budget.snapshot()['spent_usd'] == 2
+    with pytest.raises(InfraError):
+        wrapped.run(ep, time.monotonic()+1)
+    assert inner.run.call_count == 1
+
+
+def test_wrapper_preserves_model_metadata_used_by_stored_episode_rows(tmp_path):
+    wrapped, inner, _, _ = competitor(tmp_path)
+    inner.model_label = 'monarch@abc123+prototype'
+    wrapped = BudgetedCompetitor(inner, CampaignBudget(tmp_path/'other.sqlite'),
+                                 authorize=lambda: None, stop=Mock())
+    assert wrapped.model_label == 'monarch@abc123+prototype'
+
+
+def test_unknown_cost_stops_paid_work_and_does_not_treat_zero_as_free(tmp_path):
+    wrapped, inner, budget, stop = competitor(tmp_path, ArmResult(cost_usd=0, flags=['cost_missing']))
+    wrapped.run(SimpleNamespace(episode_id='a'), time.monotonic()+1)
+    assert budget.snapshot()['unknown'] == ['a']
+    stop.assert_called()
+    with pytest.raises(InfraError):
+        wrapped.run(SimpleNamespace(episode_id='b'), time.monotonic()+1)
+    assert inner.run.call_count == 1
+
+
+def test_timeout_cost_is_unknown_until_cancellation_and_billing_are_settled(tmp_path):
+    wrapped, inner, budget, stop = competitor(tmp_path)
+    exc = EpisodeTimeout('cancel requested')
+    exc.partial = ArmResult(cost_usd=1)
+    inner.run.side_effect = exc
+    with pytest.raises(EpisodeTimeout):
+        wrapped.run(SimpleNamespace(episode_id='a'), time.monotonic()+1)
+    assert budget.snapshot()['unknown'] == ['a']
+    stop.assert_called()
+
+
+def test_incomplete_child_usage_with_partial_cost_remains_unknown(tmp_path):
+    result = ArmResult(cost_usd=3, turn_log=[
+        {'authoring_events': {'complete': True, 'events': [
+            {'seq': 1, 'data': {'kind': 'section_authoring', 'status': 'started',
+                                'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+            {'seq': 2, 'data': {'kind': 'section_model_usage', 'callId': 'child-1',
+                                'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                                'usageComplete': True, 'costKnown': True}},
+            {'seq': 3, 'data': {'kind': 'section_authoring', 'status': 'started',
+                                'buildAttempt': 1, 'phase': 'author', 'sectionId': 'write', 'attempt': 0}},
+            {'seq': 4, 'data': {'kind': 'section_model_usage', 'callId': 'child-2',
+                                'buildAttempt': 1, 'phase': 'author', 'sectionId': 'write', 'attempt': 0,
+                                'usageComplete': False, 'costKnown': False}},
+        ]}},
+        {'cost': {'authoring': {'model': {'cost_usd': 3}}}},
+    ])
+    wrapped, _, budget, stop = competitor(tmp_path, result)
+    wrapped.run(SimpleNamespace(episode_id='partial'), time.monotonic()+1)
+    assert budget.snapshot()['unknown'] == ['partial']
+    stop.assert_called_once()
+
+
+def test_complete_child_usage_allows_exactly_one_settlement(tmp_path):
+    result = ArmResult(cost_usd=3, turn_log=[
+        {'authoring_events': {'complete': True, 'events': [
+            {'seq': 1, 'data': {'kind': 'section_authoring', 'status': 'started',
+                                'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+            {'seq': 2, 'data': {'kind': 'section_model_usage', 'callId': 'child-1',
+                                'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                                'usageComplete': True, 'costKnown': True}},
+        ]}},
+        {'cost': {'authoring': {'model': {'cost_usd': 3}}}},
+    ])
+    wrapped, inner, budget, _ = competitor(tmp_path, result)
+    wrapped.run(SimpleNamespace(episode_id='complete'), time.monotonic()+1)
+    assert budget.snapshot()['spent_usd'] == 3
+    with pytest.raises(InfraError):
+        wrapped.run(SimpleNamespace(episode_id='complete'), time.monotonic()+1)
+    assert inner.run.call_count == 1
+
+
+def test_durable_usage_matches_every_launch_and_allows_zero_child_current_mode():
+    assert durable_section_usage_complete([
+        {'authoring_events': {'complete': True, 'events': []}},
+    ])
+    paired = [
+        {'seq': 1, 'data': {'kind': 'section_authoring', 'status': 'started',
+                            'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+        {'seq': 2, 'data': {'kind': 'section_model_usage', 'callId': 'one',
+                            'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                            'usageComplete': True, 'costKnown': True}},
+        {'seq': 3, 'data': {'kind': 'section_authoring', 'status': 'started',
+                            'buildAttempt': 2, 'phase': 'plan', 'attempt': 0}},
+        {'seq': 4, 'data': {'kind': 'section_model_usage', 'callId': 'two',
+                            'buildAttempt': 2, 'phase': 'plan', 'attempt': 0,
+                            'usageComplete': True, 'costKnown': True}},
+    ]
+    assert durable_section_usage_complete([
+        {'authoring_events': {'complete': True, 'events': paired}},
+    ])
+
+
+def test_durable_usage_accepts_normal_progress_and_complete_section_lifecycle():
+    events = [
+        {'type': 'progress', 'data': None},
+        {'data': {'kind': 'section_authoring', 'status': 'started',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+        {'data': {'kind': 'section_model_usage', 'callId': 'planner',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                  'usageComplete': True, 'costKnown': True}},
+        {'data': {'kind': 'section_authoring', 'status': 'completed',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+        {'data': {'kind': 'section_plan', 'buildAttempt': 1, 'revision': 'v1',
+                  'sections': [{'id': 'write', 'nodeIds': ['save'], 'imports': []}]}},
+        {'data': {'kind': 'section_authoring', 'status': 'started',
+                  'buildAttempt': 1, 'phase': 'author', 'sectionId': 'write', 'attempt': 0}},
+        {'data': {'kind': 'section_model_usage', 'callId': 'writer',
+                  'buildAttempt': 1, 'phase': 'author', 'sectionId': 'write', 'attempt': 0,
+                  'usageComplete': True, 'costKnown': True}},
+        {'data': {'kind': 'section_authoring', 'status': 'completed',
+                  'buildAttempt': 1, 'phase': 'author', 'sectionId': 'write', 'attempt': 0}},
+    ]
+    for phase in ['merge', 'validate', 'build']:
+        for status in ['started', 'completed']:
+            events.append({'data': {'kind': 'section_authoring', 'buildAttempt': 1,
+                                    'phase': phase, 'status': status}})
+    events.extend([{'data': {'kind': 'section_assembly', 'buildAttempt': 1}}, {'type': 'done'}])
+    assert durable_section_usage_complete([
+        {'authoring_events': {'complete': False, 'events': events[:6]}},
+        {'authoring_events': {'complete': True, 'events': events[6:]}},
+    ])
+    assert durable_section_usage_complete([
+        {'authoring_events': {'complete': True, 'events': [{'data': None}, {'type': 'done'}]}},
+    ])
+
+
+def test_durable_usage_matches_reused_keys_across_clarification_resumes():
+    pages = []
+    for call_id in ['before-clarification', 'after-clarification']:
+        pages.append({'authoring_events': {'complete': False, 'events': [
+            {'data': {'kind': 'section_authoring', 'status': 'started',
+                      'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+            {'data': {'kind': 'section_model_usage', 'callId': call_id,
+                      'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                      'usageComplete': True, 'costKnown': True}},
+        ]}})
+    pages[-1]['authoring_events']['complete'] = True
+    assert durable_section_usage_complete(pages)
+
+
+@pytest.mark.parametrize('events', [
+    [None],
+    [{'data': []}],
+    [{'data': {'kind': 'section_authoring', 'status': 'started', 'phase': 'plan'}}],
+    [{'data': {'kind': 'section_model_usage', 'phase': 'merge'}}],
+    [
+        {'data': {'kind': 'section_model_usage', 'callId': 'early',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                  'usageComplete': True, 'costKnown': True}},
+        {'data': {'kind': 'section_authoring', 'status': 'started',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+    ],
+    [
+        *[{'data': {'kind': 'section_authoring', 'status': 'started',
+                    'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}}] * 2,
+        {'data': {'kind': 'section_model_usage', 'callId': 'only-one-receipt',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                  'usageComplete': True, 'costKnown': True}},
+    ],
+])
+def test_durable_usage_rejects_malformed_or_out_of_order_accounting(events):
+    assert not durable_section_usage_complete([
+        {'authoring_events': {'complete': True, 'events': events}},
+    ])
+
+
+@pytest.mark.parametrize('events', [
+    [{'data': {'kind': 'section_authoring', 'status': 'started',
+               'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}}],
+    [{'data': {'truncated': True}}],
+    [
+        {'data': {'kind': 'section_authoring', 'status': 'started',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0}},
+        {'data': {'kind': 'section_model_usage', 'callId': 'same',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                  'usageComplete': True, 'costKnown': True}},
+        {'data': {'kind': 'section_model_usage', 'callId': 'same',
+                  'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+                  'usageComplete': True, 'costKnown': True}},
+    ],
+    [{'data': {'kind': 'section_model_usage', 'callId': 'orphan',
+               'buildAttempt': 1, 'phase': 'plan', 'attempt': 0,
+               'usageComplete': True, 'costKnown': True}}],
+])
+def test_durable_usage_rejects_missing_truncated_duplicate_and_unmatched_evidence(events):
+    assert not durable_section_usage_complete([
+        {'authoring_events': {'complete': True, 'events': events}},
+    ])
+
+
+def test_preflight_is_rechecked_before_every_dispatch(tmp_path):
+    wrapped, inner, budget, stop = competitor(tmp_path)
+    wrapped.authorize = Mock(side_effect=PreflightError('proof withdrawn'))
+    with pytest.raises(PreflightError):
+        wrapped.run(SimpleNamespace(episode_id='a'), time.monotonic()+1)
+    inner.run.assert_not_called()
+    assert budget.snapshot()['committed_usd'] == 0
+
+
+def proof_fixture(tmp_path, manifest):
+    import hashlib
+    # Synthetic qualified manifest is only for proof-validator unit tests.
+    for task in manifest['tasks']:
+        task['readiness']['status'] = 'qualified'
+    sha = 'a' * 40
+    proof = {'manifest_sha256': manifest['manifest_sha256'], 'preview_sha': sha,
+             'preview_url': 'https://pr-123.monarch-dev.testbox.com', 'approved_by': 'unit-test-only',
+             'campaign_id': 'unit-test', 'personal_30_node_workflow_not_included': True}
+    def record(name, fields):
+        raw = json.dumps({'verified': True, 'preview_sha': sha, **fields}).encode()
+        p = tmp_path / f'{name}.json'
+        p.write_bytes(raw)
+        return {'path': p.name, 'sha256': hashlib.sha256(raw).hexdigest()}
+    proof['models'] = record('models', {'kind': 'exact_model_inventory', 'all_roles_accounted_for': True,
+                                      'models': {'test-role': {'model_id': 'unit-test-model', 'provider': 'test', 'effort': 'none'}}})
+    proof['dollar_enforcement'] = record('dollars', {'kind': 'server_dollar_enforcement', 'campaign_limit_usd': 1000,
+                                                   'development_limit_usd': 200, 'attempt_limit_usd': 12,
+                                                   'inflight_calls_included': True, 'unknown_usage_blocks': True,
+                                                   'enforced_before_model_calls': True})
+    proof['cancellation'] = record('cancel', {'kind': 'settled_cancellation', 'all_child_calls_stopped': True, 'billing_final': True})
+    proof['world'] = record('world', {'kind': 'dedicated_synthetic_world', 'per_attempt_reset': True, 'shared_accounts': False})
+    proof['graders'] = {t['id']: record(t['id'], {'kind': 'task_grader_controls', 'task_sha256': t['file_sha256'],
+                                               'positive_passed': True, 'negative_rejected': True, 'collateral_rejected': True})
+                        for t in manifest['tasks']}
+    path = tmp_path / 'proof.json'
+    path.write_text(json.dumps(proof))
+    return path, proof
+
+
+def test_validated_attestations_are_bound_to_the_exact_evidence_bytes(tmp_path):
+    manifest = build_manifest()
+    path, proof = proof_fixture(tmp_path, manifest)
+    assert validate_preflight(path, manifest) == proof
+    (tmp_path / 'dollars.json').write_text('{}')
+    with pytest.raises(PreflightError, match='hash differs'):
+        validate_preflight(path, manifest)
+
+
+@pytest.mark.parametrize('missing', ['models', 'dollar_enforcement', 'cancellation', 'world', 'graders'])
+def test_each_paid_proof_requirement_is_mandatory(tmp_path, missing):
+    manifest = build_manifest()
+    path, proof = proof_fixture(tmp_path, manifest)
+    del proof[missing]
+    path.write_text(json.dumps(proof))
+    with pytest.raises(PreflightError):
+        validate_preflight(path, manifest)
+
+
+def test_existing_orchestrator_applies_wrapper_before_any_competitor_prepare(tmp_path, monkeypatch):
+    from wb_orchestrator import orchestrator as module
+    from wb_results.store import Store
+    from wb_world.episode import load_task_file
+    task = load_task_file(ROOT/'tasks/simple.email_sf_contact_city_update.json')
+    seen = []
+    fake = SimpleNamespace(name='null', provider_key=None)
+    monkeypatch.setattr(module, 'build_arm', lambda key: fake)
+    store = Store(tmp_path/'results.sqlite')
+    orchestrator = module.Orchestrator(store, tmp_path, ['null'], 1, tmp_path, tasks=[task])
+    orchestrator.arm_wrapper = lambda inner: seen.append(inner) or inner
+    monkeypatch.setattr(orchestrator, '_run_arm_group', lambda run_id, arm, work: seen.append(('dispatched', arm)))
+    orchestrator.run('wrapper-test')
+    assert seen == [fake, ('dispatched', fake)]
+    store.close()
+
+
+def test_external_attestation_cannot_override_a_locally_rejected_grader(tmp_path):
+    manifest = build_manifest()
+    path, proof = proof_fixture(tmp_path, manifest)
+    manifest['tasks'][0]['readiness']['status'] = 'rejected'
+    with pytest.raises(PreflightError, match='Local independent'):
+        validate_preflight(path, manifest)
diff --git a/monarch-benchmark/workflowbench/tests/test_prototype_controls.py b/monarch-benchmark/workflowbench/tests/test_prototype_controls.py
new file mode 100644
index 0000000..c9cb33c
--- /dev/null
+++ b/monarch-benchmark/workflowbench/tests/test_prototype_controls.py
@@ -0,0 +1,50 @@
+import copy
+from collections import Counter
+
+from wb_orchestrator.prototype_campaign import ROOT, build_manifest, paired_schedule
+from wb_orchestrator.prototype_controls import qualify
+from wb_world.episode import load_task_file
+
+
+def test_prompt_derived_controls_do_not_read_assertions_for_answers():
+    for path in ['simple.email_sf_contact_city_update.json', 'simple.sf_opp_closed_won.json']:
+        task = load_task_file(ROOT/'tasks'/path)
+        result = qualify(task)
+        assert result['positive_control'] == 'passed'
+        assert result['collateral_control'] == 'rejected'
+        assert result['status'] == 'qualified'
+        assert result['evidence']['positive']['calls']
+        poisoned = copy.deepcopy(task)
+        for assertion in poisoned['info']['assertions']:
+            if 'value' in assertion:
+                assertion['value'] = 'poisoned answer'
+        inputs = lambda r: [(c['method'], c['url'], c['body']) for c in r['evidence']['positive']['calls']]
+        assert inputs(qualify(poisoned)) == inputs(result)
+
+
+def test_invoice_wrong_business_values_are_detected_as_grader_gap():
+    task = load_task_file(ROOT/'tasks/tier-medium/simple.invoice_airtable_slack.json')
+    result = qualify(task)
+    assert result['positive_control'] == 'passed'
+    assert result['semantic_counterexample'] == 'accepted'
+    assert result['status'] == 'rejected'
+    assert result['evidence']['wrong_invoice']['grade']['passed']
+
+
+def test_paired_schedule_is_seeded_balanced_and_preserves_every_comparison():
+    manifest = build_manifest()
+    schedule = paired_schedule(manifest['tasks'], manifest['phases'], seed=20260907)
+    assert schedule == manifest['schedule']
+    assert schedule == paired_schedule(manifest['tasks'], manifest['phases'], seed=20260907)
+    assert schedule != paired_schedule(manifest['tasks'], manifest['phases'], seed=20260908)
+    assert len(schedule) == 66
+    for phase in manifest['phases']:
+        rows = [r for r in schedule if r['phase'] == phase['name']]
+        n = len(phase['configurations'])
+        for i in range(0,len(rows), n):
+            block = rows[i:i+n]
+            assert len({(r['task_id'],r['repetition']) for r in block}) == 1
+            assert {r['configuration'] for r in block} == set(phase['configurations'])
+        for arm in phase['configurations']:
+            positions = Counter(r['position'] for r in rows if r['configuration'] == arm)
+            assert max(positions.values()) - min(positions.values()) <= 1
diff --git a/monarch-benchmark/workflowbench/wb_arms/http_shim.py b/monarch-benchmark/workflowbench/wb_arms/http_shim.py
index 50983e0..1e47cdc 100644
--- a/monarch-benchmark/workflowbench/wb_arms/http_shim.py
+++ b/monarch-benchmark/workflowbench/wb_arms/http_shim.py
@@ -29,9 +29,9 @@
 
 
 class _Server(ThreadingHTTPServer):
-    # A busy fixed port must fail loudly: on Windows the SO_REUSEADDR that
-    # HTTPServer sets lets a second bind steal a port that is already serving.
-    allow_reuse_address = False
+    # POSIX needs SO_REUSEADDR to reclaim a fixed port from closed connections.
+    # Windows keeps it off because there it can steal an address from a live server.
+    allow_reuse_address = os.name != "nt"
 
 
 class EpisodeHTTPShim:
@@ -153,6 +153,7 @@ def start(self) -> "EpisodeHTTPShim":
 
     def stop(self) -> None:
         self.httpd.shutdown()
+        self._thread.join()
         self.httpd.server_close()
 
 
diff --git a/monarch-benchmark/workflowbench/wb_arms/monarch.py b/monarch-benchmark/workflowbench/wb_arms/monarch.py
index 960da32..99818a9 100644
--- a/monarch-benchmark/workflowbench/wb_arms/monarch.py
+++ b/monarch-benchmark/workflowbench/wb_arms/monarch.py
@@ -396,6 +396,8 @@ def _attempt(self, ep: Episode, deadline: float) -> ArmResult:
             self._execute(client, ep, workflow_id, deadline, res, ids)
             return res
         finally:
+            if self.harness.builder_experiment and workflow_id is None:
+                workflow_id = ids.get("workflowId")
             if workflow_id and self.keep_workflows:
                 # `wb monarch recipes` decides keep-or-delete from the checker's
                 # verdict, which needs the snapshot this attempt has not taken yet,
@@ -429,12 +431,20 @@ def _reply(self, client, recipe_run: str, frame: dict, deadline: float) -> int |
                      deadline=deadline)
         return len(questions)
 
-    def _cancel(self, client, recipe_run: str) -> None:
+    def _cancel(self, client, recipe_run: str):
         """Best effort: a cancel that fails must not hide why the attempt ended."""
         try:
-            client.cancel(recipe_run, deadline=time.monotonic() + 30)
-        except (InfraError, MonarchRefused):
-            pass
+            return client.cancel(recipe_run, deadline=time.monotonic() + 30)
+        except (InfraError, MonarchRefused) as error:
+            return error
+
+    def _cancel_authoring_prototype(self, client, recipe_run: str, res: ArmResult) -> None:
+        response = self._cancel(client, recipe_run)
+        if isinstance(response, dict) and response.get('ok') is True:
+            res.turn_log.append({'authoring_cancel': response})
+        else:
+            raw = str(response) if isinstance(response, BaseException) else response
+            res.turn_log.append({'authoring_cancel_error': raw})
 
     def _delete(self, client, workflow_id: str) -> bool:
         """Delete one workflow; remember it for the next attempt if it will not go."""
@@ -468,7 +478,9 @@ def _author(self, client, ep, goal, deadline, res, ids) -> str | None:
         """Stream the authoring run; return the workflow id, or fill `res` and return None."""
         t0 = time.monotonic()
         recipe_run = client.start_authoring(goal, self._bench_id, deadline=deadline,
-                                            authoring_mode=self.harness.authoring_mode)
+                                            authoring_mode=self.harness.authoring_mode,
+                                            **({"experiment": self.harness.builder_experiment}
+                                               if self.harness.builder_experiment else {}))
         ids["recipeRunId"] = recipe_run
         workflow_id, questions = None, 0
         answered: set[str] = set()     # a reconnected stream replays the prompt
@@ -483,10 +495,13 @@ def _author(self, client, ep, goal, deadline, res, ids) -> str | None:
                 done = False
                 while not done:
                     try:
-                        frames = client.stream(recipe_run, deadline=deadline)
+                        frames = (client.authoring_snapshots(recipe_run, deadline=deadline)
+                                  if self.harness.builder_experiment else client.stream(recipe_run, deadline=deadline))
                         for frame in frames:
                             res.turn_log.append({"frame": frame})
                             self._note_trace(frame)
+                            if self.harness.builder_experiment and (frame.get("experiment") or {}).get("id") != self.harness.builder_experiment:
+                                raise InfraError("infra:harness_crash", "Persisted builder experiment does not match the requested competitor", retryable=False)
                             status = frame.get("status")
                             if status == "done":
                                 workflow_id = frame.get("workflowId")
@@ -507,7 +522,8 @@ def _author(self, client, ep, goal, deadline, res, ids) -> str | None:
                                 break
                             if status == "error":
                                 message = frame.get("error")
-                                self._infra = _classify_authoring_error(message)
+                                self._infra = (None if self.harness.builder_experiment
+                                               else _classify_authoring_error(message))
                                 res.termination = "agent_error"
                                 res.error = f"authoring_error: {message}"
                                 done = True
@@ -522,11 +538,16 @@ def _author(self, client, ep, goal, deadline, res, ids) -> str | None:
                                 if asked is None:   # an account prompt, or nothing to answer
                                     res.termination = "agent_error"
                                     res.error = "account_requested"
-                                    self._cancel(client, recipe_run)
+                                    if self.harness.builder_experiment:
+                                        self._cancel_authoring_prototype(client, recipe_run, res)
+                                    else:
+                                        self._cancel(client, recipe_run)
                                     done = True
                                     break
                                 questions += asked
                     except InfraError as e:
+                        if self.harness.builder_experiment:
+                            raise
                         # A 404 means the job no longer exists; reconnecting to a
                         # backend that is merely unwell is the deadline's problem.
                         if "404" not in str(e):
@@ -543,14 +564,28 @@ def _author(self, client, ep, goal, deadline, res, ids) -> str | None:
                         time.sleep(self.POLL_INTERVAL_S)
             except EpisodeTimeout as e:
                 # FR-012: nothing is left running behind a timed-out attempt.
-                self._cancel(client, recipe_run)
+                if self.harness.builder_experiment:
+                    self._cancel_authoring_prototype(client, recipe_run, res)
+                else:
+                    self._cancel(client, recipe_run)
                 raise EpisodeTimeout(f"deadline passed in the authoring phase: {e}") from e
+        except (InfraError, MonarchRefused):
+            if self.harness.builder_experiment:
+                self._cancel_authoring_prototype(client, recipe_run, res)
+            raise
         finally:
             # Recorded even on a timeout: the phase clock is the FR-010 detail
             # that says where the deadline passed.
             res.phases["authoring"] = PhaseMetrics(turns=questions,
                                                    wall_clock_s=round(time.monotonic() - t0, 4))
             res.flags.append(f"questions_asked={questions}")
+        if self.harness.builder_experiment:
+            try:
+                for page in client.authoring_event_pages(recipe_run, deadline=deadline):
+                    res.turn_log.append({"authoring_events": page})
+            except MonarchRefused as error:
+                res.turn_log.append({"authoring_events_error": {"status": error.status, "body": error.body}})
+                raise InfraError("infra:harness_crash", f"Authoring event export refused with HTTP {error.status}", retryable=False) from error
         return workflow_id
 
     def _start_run(self, client, ep, workflow_id, deadline, res) -> dict | None:
@@ -681,5 +716,13 @@ def _execute(self, client, ep, workflow_id, deadline, res, ids) -> None:
                                      f"node={out.get('errorNodeId')}")
                     return
                 time.sleep(self.POLL_INTERVAL_S)
+        except (EpisodeTimeout, InfraError, MonarchRefused):
+            if self.harness.builder_experiment and ids.get("runId"):
+                try:
+                    cancelled = client.cancel_execution(ids["runId"], deadline=time.monotonic() + 30)
+                    res.turn_log.append({"execution_cancel": cancelled})
+                except (InfraError, MonarchRefused) as error:
+                    res.turn_log.append({"execution_cancel_error": str(error)})
+            raise
         finally:
             res.phases["execution"] = PhaseMetrics(wall_clock_s=round(time.monotonic() - t0, 4))
diff --git a/monarch-benchmark/workflowbench/wb_arms/monarch_client.py b/monarch-benchmark/workflowbench/wb_arms/monarch_client.py
index b8e1221..e8d6cf4 100644
--- a/monarch-benchmark/workflowbench/wb_arms/monarch_client.py
+++ b/monarch-benchmark/workflowbench/wb_arms/monarch_client.py
@@ -108,16 +108,66 @@ def liveness(self, deadline: float | None = None) -> bool:
     # -- authoring -------------------------------------------------------------
 
     def start_authoring(self, goal: str, episode_id: str, deadline: float | None = None,
-                        authoring_mode: str | None = None) -> str:
+                        authoring_mode: str | None = None, experiment: str | None = None) -> str:
         """Start an authoring job. `authoring_mode="unattended"` asks the builder not
         to stop for questions; anything else sends today's body, with no such field."""
         body = {"goal": goal}
+        if experiment is not None:
+            body["experiment"] = experiment
         if authoring_mode == "unattended":
             body["authoring"] = "unattended"
         out = self._call("POST", "/api/workflows/recipe/runs", body,
                          headers={"x-bench-episode-id": episode_id}, deadline=deadline)
         return out["runId"]
 
+    def get_authoring(self, run_id: str, deadline: float | None = None) -> dict:
+        return self._call("GET", f"/api/workflows/recipe/runs/{run_id}", deadline=deadline)
+
+    def authoring_snapshots(self, run_id: str, deadline: float, poll_interval_s: float = 2.0) -> Iterator[dict]:
+        while True:
+            if time.monotonic() >= deadline:
+                raise EpisodeTimeout(f"deadline polling authoring run {run_id}")
+            snapshot = self.get_authoring(run_id, deadline)
+            yield snapshot
+            if snapshot.get("status") in {"done", "error"}:
+                return
+            time.sleep(min(poll_interval_s, max(0, deadline - time.monotonic())))
+
+    def authoring_event_pages(self, run_id: str, deadline: float, poll_interval_s: float = 2.0) -> Iterator[dict]:
+        """Yield raw pages until the durable terminal ledger is fully downloaded.
+
+        410 remains MonarchRefused with its original response. Existing payload
+        truncation markers are retained. No incomplete export becomes success.
+        """
+        cursor = 0
+        while True:
+            if time.monotonic() >= deadline:
+                raise EpisodeTimeout(f"deadline exporting authoring events for {run_id}")
+            page = self._call("GET", f"/api/workflows/recipe/runs/{run_id}/events?afterSeq={cursor}&limit=1000", deadline=deadline)
+            events, next_cursor = page.get("events"), page.get("nextAfterSeq")
+            more, complete = page.get("hasMore"), page.get("complete")
+            valid = (page.get("runId") == run_id and isinstance(events, list)
+                     and type(next_cursor) is int and next_cursor >= cursor
+                     and type(more) is bool and type(complete) is bool
+                     and not (more and complete) and (not more or next_cursor > cursor))
+            previous = cursor
+            if valid:
+                for event in events:
+                    seq = event.get("seq") if isinstance(event, dict) else None
+                    if type(seq) is not int or seq <= previous or seq > next_cursor:
+                        valid = False
+                        break
+                    previous = seq
+                valid = valid and next_cursor == previous
+            if not valid:
+                raise InfraError("infra:harness_crash", f"Invalid authoring event page: {page!r}", retryable=False)
+            yield page
+            cursor = next_cursor
+            if complete:
+                return
+            if not more:
+                time.sleep(min(poll_interval_s, max(0, deadline - time.monotonic())))
+
     def stream(self, run_id: str, deadline: float | None = None) -> Iterator[dict]:
         """Yield the parsed `data:` frames of the authoring stream until it closes.
 
@@ -190,6 +240,9 @@ def workflow_runs(self, workflow_id: str, deadline: float | None = None) -> list
         out = self._call("GET", f"/api/workflows/{workflow_id}/runs", deadline=deadline)
         return out.get("items") or []
 
+    def cancel_execution(self, run_id: str, deadline: float | None = None) -> dict:
+        return self._call("POST", f"/api/engine/runs/{run_id}/cancel", {}, deadline=deadline)
+
     def get_run(self, run_id: str, deadline: float | None = None) -> dict:
         return self._call("GET", f"/api/workflows/runs/{run_id}", deadline=deadline)
 
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/campaign_budget.py b/monarch-benchmark/workflowbench/wb_orchestrator/campaign_budget.py
new file mode 100644
index 0000000..788340d
--- /dev/null
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/campaign_budget.py
@@ -0,0 +1,130 @@
+"""Atomic, local campaign reservations; this does not meter or stop model calls.
+
+Callers must reserve before dispatch, dispatch only newly created reservations,
+then reconcile the complete provider cost (or None when cost is unknown).
+"""
+from __future__ import annotations
+
+import sqlite3
+from contextlib import contextmanager
+from dataclasses import dataclass
+from decimal import Decimal, InvalidOperation, ROUND_CEILING
+from pathlib import Path
+
+
+class BudgetBlocked(RuntimeError):
+    pass
+
+
+class ReservationConflict(RuntimeError):
+    pass
+
+
+def _money(value) -> int:
+    if isinstance(value, bool):
+        raise ValueError('Cost must be a non-negative finite amount')
+    try:
+        amount = Decimal(str(value))
+        if not amount.is_finite() or amount < 0:
+            raise ValueError('Cost must be a non-negative finite amount')
+        return int((amount * 1_000_000).to_integral_value(rounding=ROUND_CEILING))
+    except (InvalidOperation, TypeError) as exc:
+        raise ValueError('Invalid cost') from exc
+
+
+@dataclass(frozen=True)
+class Reservation:
+    id: str
+    category: str
+    reserved_usd: float
+    state: str
+    created: bool
+
+
+class CampaignBudget:
+    def __init__(self, path: str | Path, *, total_limit_usd=1000, development_limit_usd=200):
+        self.path = Path(path)
+        self.total = _money(total_limit_usd)
+        self.development = _money(development_limit_usd)
+        if not 0 < self.total <= _money(1000) or not 0 < self.development <= _money(200):
+            raise ValueError('Campaign limits cannot exceed $1000 total / $200 development')
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with self._transaction() as db:
+            db.execute('CREATE TABLE IF NOT EXISTS limits (singleton INTEGER PRIMARY KEY CHECK(singleton=1), total INTEGER NOT NULL, development INTEGER NOT NULL)')
+            db.execute('CREATE TABLE IF NOT EXISTS reservations (id TEXT PRIMARY KEY, category TEXT NOT NULL, reserved INTEGER NOT NULL, actual INTEGER, state TEXT NOT NULL)')
+            db.execute('INSERT OR IGNORE INTO limits VALUES (1, ?, ?)', (self.total, self.development))
+            limits = db.execute('SELECT total, development FROM limits WHERE singleton=1').fetchone()
+            if tuple(limits) != (self.total, self.development):
+                raise ReservationConflict('Existing campaign limits cannot change')
+
+    @contextmanager
+    def _transaction(self):
+        db = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+        db.row_factory = sqlite3.Row
+        try:
+            db.execute('BEGIN IMMEDIATE')
+            yield db
+            db.commit()
+        except BaseException:
+            db.rollback()
+            raise
+        finally:
+            db.close()
+
+    def reserve(self, reservation_id: str, category: str, amount_usd=None) -> Reservation:
+        if not isinstance(reservation_id, str) or not reservation_id.strip():
+            raise ValueError('A stable reservation ID is required')
+        if category not in ('development', 'measured'):
+            raise ValueError('Unknown budget category')
+        amount = _money(12 if amount_usd is None and category == 'measured' else amount_usd)
+        if amount <= 0 or (category == 'measured' and amount != _money(12)):
+            raise ValueError('Measured attempts reserve exactly $12; development reserves a positive amount')
+        with self._transaction() as db:
+            prior = db.execute('SELECT * FROM reservations WHERE id=?', (reservation_id,)).fetchone()
+            if prior:
+                if prior['category'] != category or prior['reserved'] != amount:
+                    raise ReservationConflict('Reservation ID already has different terms')
+                return Reservation(reservation_id, category, amount / 1_000_000, prior['state'], False)
+            rows = db.execute('SELECT * FROM reservations').fetchall()
+            if any(r['state'] == 'unknown' for r in rows):
+                raise BudgetBlocked('Unknown usage must be reconciled before more paid work')
+            if any(r['actual'] is not None and r['actual'] > r['reserved'] for r in rows):
+                raise BudgetBlocked('A reservation overran its limit; campaign requires review')
+            committed = sum(r['actual'] if r['actual'] is not None else r['reserved'] for r in rows)
+            dev = sum(r['actual'] if r['actual'] is not None else r['reserved']
+                      for r in rows if r['category'] == 'development')
+            if committed + amount > self.total or (category == 'development' and dev + amount > self.development):
+                raise BudgetBlocked('Campaign or development budget would be exceeded')
+            db.execute('INSERT INTO reservations VALUES (?, ?, ?, NULL, ?)',
+                       (reservation_id, category, amount, 'reserved'))
+            return Reservation(reservation_id, category, amount / 1_000_000, 'reserved', True)
+
+    def reconcile(self, reservation_id: str, actual_usd) -> None:
+        actual = None if actual_usd is None else _money(actual_usd)
+        with self._transaction() as db:
+            row = db.execute('SELECT * FROM reservations WHERE id=?', (reservation_id,)).fetchone()
+            if row is None:
+                raise ReservationConflict('No reservation exists for this cost')
+            if row['state'] == 'reconciled':
+                if row['actual'] != actual:
+                    raise ReservationConflict('This reservation already has a different final cost')
+                return
+            db.execute('UPDATE reservations SET actual=?, state=? WHERE id=?',
+                       (actual, 'unknown' if actual is None else 'reconciled', reservation_id))
+
+    def snapshot(self) -> dict:
+        with self._transaction() as db:
+            rows = db.execute('SELECT * FROM reservations ORDER BY id').fetchall()
+            spent = sum(r['actual'] or 0 for r in rows)
+            reserved = sum(r['reserved'] for r in rows if r['actual'] is None)
+            return {
+                'total_limit_usd': self.total / 1_000_000,
+                'development_limit_usd': self.development / 1_000_000,
+                'pending': [r['id'] for r in rows if r['state'] == 'reserved'],
+                'spent_usd': spent / 1_000_000,
+                'reserved_usd': reserved / 1_000_000,
+                'committed_usd': (spent + reserved) / 1_000_000,
+                'remaining_usd': (self.total - spent - reserved) / 1_000_000,
+                'unknown': [r['id'] for r in rows if r['state'] == 'unknown'],
+                'overruns': [r['id'] for r in rows if r['actual'] is not None and r['actual'] > r['reserved']],
+            }
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/config.py b/monarch-benchmark/workflowbench/wb_orchestrator/config.py
index abd2352..8a18df1 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/config.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/config.py
@@ -131,6 +131,7 @@ class Harness:
     price_table: str | None = None
     monarch_repo: str | None = None
     modes: list[str] = field(default_factory=list)
+    builder_experiment: str | None = None
     authoring_mode: str = "interactive"   # "unattended": the builder does not stop to ask
 
 
@@ -333,7 +334,7 @@ def load_model(path) -> Model:
     "monarch": (("base_url", "credential_env", "login_email", "login_password_env", "fd_url",
                  "shim_port", "langfuse_url", "langfuse_public_key_env", "langfuse_secret_key_env",
                  "price_table", "monarch_repo", "modes"),
-                ("shim_public_host", "shim_public_url", "fd_api_key_env", "authoring_mode")),
+                ("shim_public_host", "shim_public_url", "fd_api_key_env", "authoring_mode", "builder_experiment")),
 }
 
 
@@ -418,6 +419,7 @@ def load_harness(path) -> Harness:
         price_table=c.get("price_table", str),
         monarch_repo=c.get("monarch_repo", str),
         modes=c.str_list("modes", enum=MODES, default=[]),
+        builder_experiment=c.get("builder_experiment", str, enum={"current", "compiled", "sections-serial", "sections-parallel", "sections-parallel-current"}),
         authoring_mode=c.get("authoring_mode", str, default=Harness.authoring_mode,
                              enum=AUTHORING_MODES),
     )
@@ -588,6 +590,8 @@ def _hashed_harness(h: Harness) -> dict:
         # ponytail: the default is dropped so runs frozen before this key stay
         # regradable; only asking for the unattended builder moves the hash.
         del d["authoring_mode"]
+    if d.get("builder_experiment") is None:
+        d.pop("builder_experiment", None)
     return d
 
 
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py b/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py
index dcb735a..ca02994 100644
--- a/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/orchestrator.py
@@ -191,6 +191,7 @@ def __init__(self, store: Store, suite_dir: str | Path, arms: list[str], k: int,
         self._thread_errors: list[BaseException] = []
         self._spent = 0.0            # cumulative cost_usd, carried over on resume
         self._stop_reason: str | None = None
+        self.arm_wrapper = None
 
     def _config(self) -> dict:
         if self.run_config:
@@ -265,6 +266,8 @@ def _execute(self, run_id: str, skip: set[tuple[str, str, int]]) -> None:
         arms = ([build_arm_for(c, self.run_config) for c in self.run_config.competitors]
                 if self.run_config
                 else [build_arm(k) for k in self.arm_keys])
+        if self.arm_wrapper is not None:
+            arms = [self.arm_wrapper(arm) for arm in arms]
         threads = []
         for arm in arms:
             work = [(task, trial) for task in self.tasks for trial in range(self.k)
@@ -396,6 +399,8 @@ def _run_episode(self, run_id: str, arm, task: dict, trial: int) -> bool:
                               "cost_usd", "turns", "tool_calls"):
                         setattr(acc, f, getattr(acc, f) + getattr(partial, f))
                     acc.turn_log.extend(partial.turn_log)
+                    acc.flags.extend(partial.flags)
+                    acc.phases.update(partial.phases)
                 if not e.retryable or attempt >= MAX_INFRA_RETRIES:
                     break
                 attempt += 1
@@ -410,12 +415,15 @@ def _run_episode(self, run_id: str, arm, task: dict, trial: int) -> bool:
                 termination, error = "agent_error", str(e)
                 break
 
-        if acc.tokens_prompt or acc.cost_usd:
+        if (acc.tokens_prompt or acc.cost_usd or acc.turn_log or acc.flags or acc.phases):
             for f in ("tokens_prompt", "tokens_cached", "tokens_cache_write", "tokens_output",
                       "cost_usd", "turns", "tool_calls"):
                 setattr(result, f, getattr(result, f) + getattr(acc, f))
             result.turn_log = acc.turn_log + result.turn_log
-            result.flags.append("spend_includes_failed_attempts")
+            result.flags = acc.flags + result.flags
+            result.phases = {**acc.phases, **result.phases}
+            if acc.tokens_prompt or acc.cost_usd:
+                result.flags.append("spend_includes_failed_attempts")
 
         # SNAPSHOT1 + GRADE + RECORD always run, whatever ARM_RUN did. A crash
         # in this stage records an infra:harness_crash row rather than losing
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/prototype_campaign.py b/monarch-benchmark/workflowbench/wb_orchestrator/prototype_campaign.py
new file mode 100644
index 0000000..b78e018
--- /dev/null
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/prototype_campaign.py
@@ -0,0 +1,382 @@
+"""Keyless campaign planning; paid dispatch requires separately supplied proof."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import random
+import shutil
+import subprocess
+from collections import Counter
+from dataclasses import asdict, replace
+from pathlib import Path
+
+from grader.grade import grade
+from wb_orchestrator.prototype_controls import qualify
+from wb_arms.api_loop import InfraError
+from wb_orchestrator.campaign_budget import CampaignBudget, BudgetBlocked
+from wb_world.episode import Episode, contract_hash, load_task_file
+
+ROOT = Path(__file__).resolve().parents[1]
+CONFIGURATIONS = ['current', 'compiled', 'sections-serial', 'sections-parallel', 'sections-parallel-current']
+SELECTION = {
+    'development': [
+        ('tasks/simple.email_sf_contact_city_update.json', 'single change baseline'),
+        ('tasks/tier-medium/simple.invoice_airtable_slack.json', 'multiple products'),
+        ('tasks/tier-simple/sales.update_contact_phone.json', 'batch entity matching'),
+        ('tasks/tier-medium/support.freshdesk_auto_merge.json', 'duplicate handling'),
+        ('tasks/tier-complex/support.intercom_freshdesk_escalation.json', 'large policy and cross-product case'),
+        ('tasks/tier-complex/sales.zoom_recording_distribution.json', 'multi-recipient policy and routing'),
+    ],
+    'holdout': [
+        ('tasks/simple.sf_opp_closed_won.json', 'single change with coupled state'),
+        ('tasks/tier-medium/operations.invoice_shipping_trigger.json', 'conditional fulfillment'),
+        ('tasks/random-10/support.reamaze_cross_platform_dedup.json', 'cross-product deduplication'),
+        ('tasks/tier-medium/hr.comp_adjustment_batch.json', 'batch eligibility and notifications'),
+    ],
+}
+
+
+class PreflightError(RuntimeError):
+    pass
+
+
+def durable_section_usage_complete(turn_log: list[dict]) -> bool:
+    pages = [record['authoring_events'] for record in turn_log
+             if isinstance(record, dict) and isinstance(record.get('authoring_events'), dict)]
+    if not pages or pages[-1].get('complete') is not True:
+        return False
+    pending, call_ids = Counter(), set()
+    for page in pages:
+        events = page.get('events')
+        if not isinstance(events, list):
+            return False
+        for event in events:
+            if not isinstance(event, dict):
+                return False
+            data = event.get('data')
+            if data is None:
+                continue
+            if not isinstance(data, dict) or data.get('truncated') is True:
+                return False
+            kind = data.get('kind')
+            if kind not in ('section_authoring', 'section_model_usage'):
+                continue
+            if kind == 'section_authoring' and data.get('phase') in ('merge', 'validate', 'build'):
+                continue
+            key = (data.get('buildAttempt'), data.get('phase'),
+                   data.get('sectionId'), data.get('attempt'))
+            valid_key = (type(key[0]) is int and key[0] > 0
+                         and key[1] in ('plan', 'author')
+                         and (key[1] == 'plan' and key[2] is None
+                              or key[1] == 'author' and isinstance(key[2], str) and bool(key[2]))
+                         and type(key[3]) is int and key[3] >= 0)
+            if not valid_key:
+                return False
+            if kind == 'section_authoring':
+                if data.get('status') != 'started':
+                    continue
+                # A clarification resume can reuse the same per-turn build key.
+                pending[key] += 1
+                continue
+            call_id = data.get('callId')
+            if (not isinstance(call_id, str) or not call_id or call_id in call_ids
+                    or pending[key] == 0 or data.get('usageComplete') is not True
+                    or data.get('costKnown') is not True):
+                return False
+            call_ids.add(call_id)
+            pending[key] -= 1
+    return not any(pending.values())
+
+
+def digest(value) -> str:
+    return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+
+
+def _control(task, action=None) -> dict:
+    episode = Episode(task, episode_id='keyless-readiness')
+    if action:
+        action.run(episode)
+    final = episode.finish()
+    return {**grade(task, episode.snapshot0, final), 'snapshot_sha256': digest(final)}
+
+
+def paired_schedule(tasks, phases, seed=20260907):
+    rng = random.Random(seed)
+    order = []
+    for phase in phases:
+        blocks = [(task['id'], repetition) for task in tasks if task['split'] == phase['name']
+                  for repetition in range(1, phase['repetitions'] + 1)]
+        rng.shuffle(blocks)
+        arms = list(phase['configurations'])
+        rng.shuffle(arms)
+        for block, (task_id, repetition) in enumerate(blocks):
+            rotated = arms[block % len(arms):] + arms[:block % len(arms)]
+            for position, setting in enumerate(rotated):
+                order.append({'phase': phase['name'], 'task_id': task_id, 'repetition': repetition,
+                              'configuration': setting, 'position': position})
+    return order
+
+
+def build_manifest(root: Path = ROOT) -> dict:
+    tasks = []
+    for split, selection in SELECTION.items():
+        for relative, purpose in selection:
+            path = root / relative
+            task = load_task_file(path)
+            info = task.get('info', {})
+            readiness = {
+                'assertions': len(info.get('assertions', [])),
+                'expected_changes': len(info.get('expected_changes', [])),
+                'source_contract_matches': task.get('contract_sha256') in (None, contract_hash(task)),
+                'negative_control': 'not_checked',
+                'positive_control': 'not_checked',
+                'collateral_control': 'not_checked',
+            }
+            try:
+                readiness['negative_control'] = 'accepted' if _control(task)['passed'] else 'rejected'
+                readiness.update({key: value for key, value in qualify(task).items() if key != 'evidence'})
+            except Exception as error:
+                readiness['error'] = f'{type(error).__name__}: {error}'
+            tasks.append({
+                'id': task['task'], 'path': relative, 'split': split, 'purpose': purpose,
+                'file_sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+                'contract_hash': contract_hash(task), 'readiness': readiness,
+            })
+    phases = [
+        {'name': 'development', 'configurations': CONFIGURATIONS, 'tasks': 6, 'repetitions': 1, 'attempts': 30},
+        {'name': 'holdout', 'configurations': ['current', 'sections-serial', 'sections-parallel'], 'tasks': 4, 'repetitions': 3, 'attempts': 36},
+    ]
+    sources = ['grader/grade.py', 'grader/invariant.py', 'wb_world/episode.py',
+               'runner/arms.py', 'wb_orchestrator/prototype_controls.py', 'wb_orchestrator/prototype_campaign.py',
+               'wb_orchestrator/campaign_budget.py', 'wb_orchestrator/orchestrator.py',
+               'wb_orchestrator/config.py', 'wb_arms/monarch.py', 'wb_arms/monarch_client.py']
+    source_hashes = {name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in sources}
+    try:
+        vendor_revision = subprocess.check_output(['git', '-C', str(ROOT / 'vendor/automation-bench'), 'rev-parse', 'HEAD'], text=True).strip()
+    except (OSError, subprocess.CalledProcessError):
+        vendor_revision = None
+    manifest = {
+        'source_sha256': source_hashes, 'automationbench_revision': vendor_revision,
+        'version': 1, 'tasks': tasks, 'phases': phases, 'attempts': 66,
+        'schedule_seed': 20260907, 'schedule': paired_schedule(tasks, phases),
+        'reservation_estimate_usd': 792, 'development_allowance_usd': 200,
+        'maximum_combined_allocation_usd': 992, 'campaign_limit_usd': 1000,
+        'attempt_limit_usd': 12, 'automatic_retries': 0, 'missing': ['personal_30_node_workflow'],
+        'qualification': 'Candidate fixtures, not evidence of 30-node authoring or representative pilot coverage. Readiness records distinguish qualified, rejected, and pending independent controls; all must qualify before paid dispatch.',
+        'paid_prerequisites': ['approved_manifest', 'exact_model_inventory', 'server_dollar_enforcement',
+                               'settled_cancellation', 'dedicated_world', 'per_task_grader_evidence'],
+    }
+    manifest['manifest_sha256'] = digest(manifest)
+    return manifest
+
+
+def _evidence(proof_path: Path, reference: dict, kind: str, preview_sha: str) -> dict:
+    if not isinstance(reference, dict) or not reference.get('path') or not reference.get('sha256'):
+        raise PreflightError(f'Missing {kind} evidence reference')
+    path = proof_path.parent / reference['path']
+    try:
+        raw = path.read_bytes()
+        record = json.loads(raw)
+    except (OSError, ValueError) as error:
+        raise PreflightError(f'Cannot read {kind} evidence') from error
+    if not isinstance(record, dict):
+        raise PreflightError(f'{kind} evidence must be an object')
+    if hashlib.sha256(raw).hexdigest() != reference['sha256']:
+        raise PreflightError(f'{kind} evidence hash differs')
+    if record.get('kind') != kind or record.get('verified') is not True or record.get('preview_sha') != preview_sha:
+        raise PreflightError(f'{kind} evidence is not verified against this preview')
+    return record
+
+
+def validate_preflight(path: str | Path | None, manifest: dict) -> dict:
+    if path is None:
+        raise PreflightError('Paid preflight proof is required; dollar enforcement is currently unverified')
+    path = Path(path)
+    try:
+        proof = json.loads(path.read_text())
+    except (OSError, ValueError) as error:
+        raise PreflightError('Cannot read paid preflight proof') from error
+    if not isinstance(proof, dict):
+        raise PreflightError('Paid preflight proof must be an object')
+    if proof.get('manifest_sha256') != manifest['manifest_sha256']:
+        raise PreflightError('Approved manifest differs from the current task/configuration manifest')
+    if not manifest.get('automationbench_revision'):
+        raise PreflightError('The AutomationBench source revision must be established')
+    if not proof.get('approved_by') or not re.fullmatch(r'[A-Za-z0-9._-]+', proof.get('campaign_id', '')):
+        raise PreflightError('Named campaign approval is required')
+    if not re.fullmatch(r'https://pr-[0-9]+\.monarch-dev\.testbox\.com', proof.get('preview_url', '')):
+        raise PreflightError('A dedicated Monarch PR preview URL is required')
+    sha = proof.get('preview_sha', '')
+    if not re.fullmatch(r'[0-9a-f]{40}', sha):
+        raise PreflightError('An exact preview commit is required')
+    if proof.get('personal_30_node_workflow_not_included') is not True:
+        raise PreflightError('Approval must acknowledge the missing personal 30-node workflow')
+    model = _evidence(path, proof.get('models'), 'exact_model_inventory', sha)
+    models = model.get('models')
+    if model.get('all_roles_accounted_for') is not True or not isinstance(models, dict) or not models:
+        raise PreflightError('A complete configured model inventory is required')
+    for role, configured in models.items():
+        if (not role or not isinstance(configured, dict) or not configured.get('model_id')
+                or not configured.get('provider') or 'effort' not in configured):
+            raise PreflightError('Every model role must name exact provider, model, and effort')
+    dollars = _evidence(path, proof.get('dollar_enforcement'), 'server_dollar_enforcement', sha)
+    if not (dollars.get('campaign_limit_usd') == 1000 and dollars.get('development_limit_usd') == 200
+            and dollars.get('attempt_limit_usd') == 12 and dollars.get('inflight_calls_included') is True
+            and dollars.get('unknown_usage_blocks') is True and dollars.get('enforced_before_model_calls') is True):
+        raise PreflightError('Server dollar enforcement is not established')
+    cancel = _evidence(path, proof.get('cancellation'), 'settled_cancellation', sha)
+    if cancel.get('all_child_calls_stopped') is not True or cancel.get('billing_final') is not True:
+        raise PreflightError('Cancellation must settle all child calls and final billing')
+    world = _evidence(path, proof.get('world'), 'dedicated_synthetic_world', sha)
+    if world.get('per_attempt_reset') is not True or world.get('shared_accounts') is not False:
+        raise PreflightError('A dedicated resettable world is required')
+    for task in manifest['tasks']:
+        if task['readiness'].get('status') != 'qualified':
+            raise PreflightError(f"Local independent grader controls are not qualified: {task['id']}")
+        if not task['readiness']['source_contract_matches'] or not task['readiness']['assertions'] or not task['readiness']['expected_changes']:
+            raise PreflightError(f"Task source/approval contract is not ready: {task['id']}")
+        review = _evidence(path, (proof.get('graders') or {}).get(task['id']), 'task_grader_controls', sha)
+        if (review.get('task_sha256') != task['file_sha256'] or review.get('positive_passed') is not True
+                or review.get('negative_rejected') is not True or review.get('collateral_rejected') is not True):
+            raise PreflightError(f"Grader controls are not verified: {task['id']}")
+    return proof
+
+
+class BudgetedCompetitor:
+    def __init__(self, inner, budget: CampaignBudget, *, authorize, stop):
+        self.inner, self.budget, self.authorize, self.stop = inner, budget, authorize, stop
+        self.name, self.provider_key = inner.name, inner.provider_key
+        self.model_label = getattr(inner, 'model_label', None)
+
+    def prepare(self):
+        self.authorize()
+        if hasattr(self.inner, 'prepare'):
+            self.inner.prepare()
+
+    def run(self, episode, deadline=None):
+        try:
+            self.authorize()
+            reservation = self.budget.reserve(episode.episode_id, 'measured')
+            if not reservation.created:
+                raise BudgetBlocked('Attempt already reserved; inspect its prior state before recovery')
+        except (BudgetBlocked, PreflightError) as error:
+            self.stop()
+            if isinstance(error, PreflightError):
+                raise
+            raise InfraError('infra:harness_crash', str(error), retryable=False) from error
+        try:
+            result = self.inner.run(episode, deadline=deadline)
+        except BaseException as error:
+            self.budget.reconcile(episode.episode_id, None)
+            self.stop()
+            if isinstance(error, InfraError):
+                error.retryable = False
+            raise
+        unknown = ('cost_missing' in result.flags or result.termination == 'timeout'
+                   or not durable_section_usage_complete(result.turn_log)
+                   or any('execution_cancel_error' in record
+                          or 'authoring_cancel_error' in record
+                          for record in result.turn_log))
+        try:
+            self.budget.reconcile(episode.episode_id, None if unknown else result.cost_usd)
+        except BaseException:
+            self.budget.reconcile(episode.episode_id, None)
+            self.stop()
+            raise
+        if unknown or self.budget.snapshot()['overruns']:
+            self.stop()
+        return result
+
+
+def execute(manifest: dict, proof_path: Path, *, product: Path, plan: Path, budget_path: Path, output: Path):
+    proof = validate_preflight(proof_path, manifest)
+    from wb_orchestrator import config
+    from wb_orchestrator.monarch_setup import expand, public_front_door_url
+    from wb_orchestrator.orchestrator import Orchestrator
+    from wb_results.store import Store
+    base = config.resolve(product, plan)
+    monarch = [c for c in base.competitors if c.harness.kind == 'monarch']
+    if len(monarch) != 1 or base.plan.mode != 'create-run':
+        raise PreflightError('Base plan must contain exactly one create-and-run Monarch competitor')
+    competitor = monarch[0]
+    if expand(competitor.harness.base_url, os.environ, 'base_url') != proof['preview_url']:
+        raise PreflightError('Configured Monarch URL differs from approved preview')
+    models = _evidence(proof_path, proof['models'], 'exact_model_inventory', proof['preview_sha'])
+    if models.get('resolved_competitor_sha256') != digest(asdict(competitor)):
+        raise PreflightError('Model evidence differs from resolved competitor model/effort/harness settings')
+    world = _evidence(proof_path, proof['world'], 'dedicated_synthetic_world', proof['preview_sha'])
+    if public_front_door_url(competitor.harness, os.environ) != world.get('front_door_url'):
+        raise PreflightError('Configured fixture front door differs from isolation evidence')
+    repo = config.from_workflowbench(competitor.harness.monarch_repo, base.config_dir)
+    try:
+        checkout_sha = subprocess.check_output(['git', '-C', str(repo), 'rev-parse', 'HEAD'], text=True).strip()
+    except (OSError, subprocess.CalledProcessError) as error:
+        raise PreflightError('Cannot establish the configured Monarch checkout revision') from error
+    if checkout_sha != proof['preview_sha']:
+        raise PreflightError('Configured Monarch checkout differs from approved preview commit')
+    # Validation and read-only configuration resolution happen before either DB exists.
+    budget = CampaignBudget(budget_path)
+    output.mkdir(parents=True, exist_ok=True)
+    (output / 'campaign-manifest.json').write_text(json.dumps(manifest, indent=2, sort_keys=True))
+    (output / 'preflight-proof.json').write_text(json.dumps(proof, indent=2, sort_keys=True))
+    store = Store(output / 'results.sqlite')
+    def authorize():
+        refreshed = build_manifest()
+        if refreshed['manifest_sha256'] != manifest['manifest_sha256']:
+            raise PreflightError('Task files changed after campaign approval')
+        if validate_preflight(proof_path, refreshed) != proof:
+            raise PreflightError('Paid proof changed during the campaign')
+        if subprocess.check_output(['git', '-C', str(repo), 'rev-parse', 'HEAD'], text=True).strip() != proof['preview_sha']:
+            raise PreflightError('Configured Monarch checkout changed during the campaign')
+    try:
+        frozen_tasks = {}
+        frozen_dir = (output / 'tasks').resolve()
+        frozen_dir.mkdir(parents=True, exist_ok=True)
+        for task in manifest['tasks']:
+            destination = frozen_dir / Path(task['path']).name
+            shutil.copyfile(ROOT / task['path'], destination)
+            frozen_tasks[task['id']] = load_task_file(destination)
+        for index, entry in enumerate(manifest['schedule']):
+            setting = entry['configuration']
+            configured = replace(competitor, harness=replace(competitor.harness, builder_experiment=setting))
+            run_plan = replace(base.plan, name=f"prototype-{index:03d}-{setting}", tasks=str(frozen_dir), repetitions=1,
+                               retry_on_fail=0, concurrency=1, timeout_s=1200,
+                               approved_by=proof['approved_by'], cost_ceiling_usd=1000)
+            run_config = replace(base, plan=run_plan, competitors=[configured], tasks=[frozen_tasks[entry['task_id']]], tasks_dir=str(frozen_dir),
+                                 harnesses={configured.harness.name: configured.harness}, excluded_tasks={})
+            orchestrator = Orchestrator.from_config(store, run_config, output)
+            orchestrator.arm_wrapper = lambda inner: BudgetedCompetitor(inner, budget, authorize=authorize, stop=orchestrator._abort.set)
+            orchestrator.run(f"{proof['campaign_id']}-{index:03d}-{entry['phase']}-{setting}")
+            if orchestrator._abort.is_set():
+                raise PreflightError('Campaign stopped after an unsettled or over-budget attempt')
+    finally:
+        store.close()
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument('--dry-plan', action='store_true', help='Emit a keyless plan without budget mutations (default)')
+    mode.add_argument('--execute', action='store_true', help='Requires verified paid preflight proof')
+    parser.add_argument('--proof', type=Path)
+    parser.add_argument('--budget', type=Path, default=Path('prototype-campaign-budget.sqlite'))
+    parser.add_argument('--output', type=Path, default=Path('prototype-campaign-results'))
+    parser.add_argument('--product', type=Path, default=ROOT / 'config/products/simulated-apps.yaml')
+    parser.add_argument('--plan', type=Path, default=ROOT / 'config/plans/pilot-monarch-single.yaml')
+    args = parser.parse_args(argv)
+    manifest = build_manifest()
+    if not args.execute:
+        print(json.dumps(manifest, indent=2, sort_keys=True))
+        return 0
+    execute(manifest, args.proof, product=args.product, plan=args.plan, budget_path=args.budget, output=args.output)
+    return 0
+
+
+if __name__ == '__main__':
+    try:
+        raise SystemExit(main())
+    except PreflightError as error:
+        raise SystemExit(f'Paid campaign blocked: {error}') from error
diff --git a/monarch-benchmark/workflowbench/wb_orchestrator/prototype_controls.py b/monarch-benchmark/workflowbench/wb_orchestrator/prototype_controls.py
new file mode 100644
index 0000000..a141e61
--- /dev/null
+++ b/monarch-benchmark/workflowbench/wb_orchestrator/prototype_controls.py
@@ -0,0 +1,94 @@
+"""Grader-only reference trajectories, independently transcribed from fixture requests.
+
+These are qualification checks, never tools or input provided to a competitor.
+Task contracts stay frozen; a counterexample rejects a candidate for review.
+"""
+import hashlib
+import json
+
+from grader.grade import grade
+from wb_world.episode import Episode
+
+SF = 'https://yourinstance.salesforce.com/services/data/v61.0/sobjects'
+PENDING = {
+    'sales.update_contact_phone': 'Implement HR batch verification, cancellation, duplicate-contact matching, and note reference; falsify unrelated contact-field edits permitted by broad expected_changes.',
+    'support.freshdesk_auto_merge': 'Independently implement ss_merge_rules/ws_rules with same-requester restriction; falsify cross-requester merges and unrelated ticket changes.',
+    'support.intercom_freshdesk_escalation': 'Independently apply ss_escalation_config, then check destinations, confirmation replies, exclusions, and untouched conversations.',
+    'sales.zoom_recording_distribution': 'Independently resolve yesterday completed meetings and distribution policy; falsify wrong audiences, excluded meetings, and duplicate sends.',
+    'operations.invoice_shipping_trigger': 'Independently calculate only -EXP line totals and counts; verify Monday fields and warehouse email, then falsify non-expedited inclusion.',
+    'support.reamaze_cross_platform_dedup': 'Independently join customer aliases and issue identity; verify reciprocal notes, close state and log, then falsify unrelated-customer closure.',
+    'hr.comp_adjustment_batch': 'Independently resolve cleared rows and current procedures; verify computed raises and employee/manager notifications, then falsify ineligible-row changes.',
+}
+
+
+def _run(task, actions):
+    episode = Episode(task, episode_id='keyless-independent-control')
+    calls = []
+    for method, url, body in actions:
+        response = episode.api_fetch(method, url, body=json.dumps(body) if body is not None else None)
+        calls.append({'method': method, 'url': url, 'body': body, 'response': response})
+    final = episode.finish()
+    return {'grade': grade(task, episode.snapshot0, final), 'calls': calls,
+            'snapshot0': episode.snapshot0, 'snapshot1': final,
+            'snapshot_sha256': hashlib.sha256(json.dumps(final, sort_keys=True).encode()).hexdigest()}
+
+
+def qualify(task):
+    name = task['task']
+    if name in PENDING:
+        return {'status': 'pending', 'positive_control': 'pending_independent_reference',
+                'collateral_control': 'pending_independent_reference', 'remaining': PENDING[name]}
+    if name == 'simple.email_sf_contact_city_update':
+        # Source: Lisa's msg_3004 relocation email, matched contact 003004.
+        actions = [('GET', 'https://gmail.googleapis.com/gmail/v1/users/me/messages/msg_3004', None),
+                   ('PATCH', f'{SF}/Contact/003004', {'MailingCity': 'Denver'})]
+        collateral = ('PATCH', f'{SF}/Contact/003004', {'Phone': '+1-555-9999'})
+    elif name == 'simple.sf_opp_closed_won':
+        actions = [('PATCH', f'{SF}/Opportunity/006001', {'StageName': 'Closed Won'})]
+        collateral = ('PATCH', f'{SF}/Opportunity/006001', {'Amount': 1})
+    elif name == 'simple.invoice_airtable_slack':
+        # Source: msg_3192, CloudHost invoice CH-2026-0089, $4,500.00.
+        actions = [('GET', 'https://gmail.googleapis.com/gmail/v1/users/me/messages/msg_3192', None),
+                   ('POST', 'https://api.airtable.com/v0/base_finance/Invoices', {'fields': {'Vendor': 'CloudHost', 'Amount': 4500}}),
+                   ('POST', 'https://slack.com/api/chat.postMessage', {'channel': 'CFIN01', 'text': 'CloudHost invoice CH-2026-0089: $4,500.00'})]
+        collateral = ('POST', 'https://api.airtable.com/v0/base_finance/Unrequested', {'fields': {'Unrequested': True}})
+    else:
+        return {'status': 'pending', 'remaining': 'No independent reference implemented'}
+    positive = _run(task, actions)
+    damaged = _run(task, actions + [collateral])
+    result = {'positive_control': 'passed' if positive['grade']['passed'] else 'failed',
+              'collateral_control': 'accepted' if damaged['grade']['passed'] else 'rejected',
+              'evidence': {'positive': positive, 'collateral': damaged}}
+    result['status'] = 'qualified' if positive['grade']['passed'] and not damaged['grade']['passed'] else 'rejected'
+    if name == 'simple.invoice_airtable_slack':
+        wrong = list(actions)
+        wrong[1] = (actions[1][0], actions[1][1], {'fields': {'Vendor': 'Wrong vendor', 'Amount': 1}})
+        evidence = _run(task, wrong)
+        result['evidence']['wrong_invoice'] = evidence
+        result['semantic_counterexample'] = 'accepted' if evidence['grade']['passed'] else 'rejected'
+        if evidence['grade']['passed']:
+            result['status'] = 'rejected'
+            result['remaining'] = 'Frozen grader accepts wrong vendor and amount; review and approve stronger assertions before selecting this task.'
+    return result
+
+
+def main():
+    import argparse
+    from pathlib import Path
+    from wb_orchestrator.prototype_campaign import ROOT, SELECTION
+    from wb_world.episode import load_task_file
+
+    parser = argparse.ArgumentParser(description='Run independent grader controls without model calls')
+    parser.add_argument('--output', type=Path, required=True)
+    args = parser.parse_args()
+    results = {}
+    for selections in SELECTION.values():
+        for relative, _ in selections:
+            task = load_task_file(ROOT / relative)
+            results[task['task']] = {'task_sha256': hashlib.sha256((ROOT / relative).read_bytes()).hexdigest(), **qualify(task)}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(results, indent=2, sort_keys=True))
+
+
+if __name__ == '__main__':
+    main()