Merge PR #1044: preserve Cowork budget and interrupted work

Integrate the reviewed Cowork changes with the current development branch.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>

devtools/benchmarks/cowork_bench/METHODOLOGY.md

@@ -98,17 +98,25 @@ infrastructure retries. Concurrent ownership is locked. A changed key, changed
 ceiling, decreased usage counter or unsettled prior run requires reconciliation.
 Unrelated spending on the same key counts conservatively toward the campaign.
 
-For the owner-approved example campaign, the requested total ceiling is **$1000**
-and an invocation's default spending bound is **$150**. The supervisor preserves
-an in-flight allowance equal to the larger of the configured reserve (default
-$100) and concurrency times the per-task bound (default $25). It stops when the
+The qualification example uses a total ceiling of **$1000** and an invocation's
+default spending bound of **$150**. The supervisor preserves
+the explicit nonnegative `--budget-reserve-usd` allowance (default $100) for
+unsettled provider charges. It is independent of concurrency and of the per-task
+lifetime bound (default $25), which remains unchanged. Select the allowance for
+billing delay and work actually in flight; multiplying full task budgets would
+prematurely stop a 32-task, $2000 campaign at $1200 spent. With a $100 reserve,
+the campaign stop boundary is $1900 spent instead. The supervisor stops when the
 campaign remainder reaches that allowance, when the invocation bound is reached,
-or when the meter or disk reserve becomes unavailable. These are configurable
-operator bounds. Billing can be delayed and paid calls may already be in flight;
+or when the meter or disk reserve becomes unavailable. A suspect meter read
+gets at most three observations requesting cache revalidation within one shared
+15-second window; only a finite, nonnegative, nondecreasing value may update the campaign. Rejected
+values and errors remain in the monitor and unit log, including after shutdown.
+This confirmation does not rebase spending or ignore known budget exhaustion.
+These are configurable operator bounds. Billing can be delayed and paid calls may already be in flight;
 the monitor is **not a provider-enforced hard dollar cap**.
 
-Start qualification at concurrency 1; the approved campaign allows at most 2
-after measuring resource use. If smoke projects the full dataset above the
+Start qualification at concurrency 1 and choose full-run concurrency after
+measuring resource use. If smoke projects the full dataset above the
 remaining campaign budget, pause for an owner decision rather than changing the
 model, effort, configuration or budget. Reconcile delayed charges before another
 paid phase. Preserve every run in a new directory outside the source and live
@@ -158,7 +166,10 @@ unknown. `llm_usage` amounts are compatibility accounting evidence; the campaign
 meter is the spending check. A billing provider such as `openrouter` does not
 identify the upstream endpoint. `response_provider` observations are reported
 only when present, with incomplete coverage disclosed; successful-call endpoint
-evidence may be unavailable in the copied logs.
+evidence may be unavailable in the copied logs. Selected scrubbed logs are
+checkpointed before each task-status request and at finalization; a slow request
+can delay the next checkpoint. An interrupted run may therefore retain partial
+evidence, while raw request/response blobs remain unexported.
 
 A result report therefore needs the exact seed and image/benchmark pins, selected
 IDs, applied settings, official evaluator outputs, complete denominator,

devtools/benchmarks/cowork_bench/README.md

@@ -50,8 +50,8 @@ docker pull postgres:15
 Start with one task and concurrency 1. Choose the task IDs from the pinned
 benchmark, then expand to a representative smoke list only after the tool
 inventory and first task work. A smoke subset is not the 496-task benchmark
-score. The campaign described in the methodology permits concurrency 2 only
-after measuring the first task's resource use.
+score. Choose full-run concurrency after measuring the first task's resource
+use and confirming the campaign budget.
 
 Create the storage directory first. Every invocation needs a **new** run root,
 including dry runs and infrastructure retries. The following dry run builds the
@@ -68,8 +68,7 @@ python3 "$COWORK_SEED/devtools/benchmarks/cowork_bench/run_cowork_bench.py" \
 
 For the first paid qualification, use a new root, omit `--dry-run`, and add the
 shared campaign record and the agreed spending limits. The following $1000
-campaign is an example of the owner-approved campaign, not a universal spending
-policy:
+campaign is a qualification example, not a universal spending policy:
 
 ```bash
 COWORK_RUN="$COWORK_STORAGE/smoke-$(date +%s)"
@@ -89,14 +88,36 @@ Reuse the **same campaign file and key** for all paid phases. If earlier probes
 predate the first campaign baseline, account for them once with
 `--prior-spend-usd` when creating the campaign.
 
+`--budget-reserve-usd` is an explicit nonnegative allowance for delayed/in-flight
+provider charges, independent of concurrency. Per-task lifetime limits remain
+separate: 32 tasks with a $25 task limit do not automatically reserve $800.
+Choose the reserve for the expected outstanding charges; delayed billing means
+it is not a provider-enforced spending cap.
+
 ## Monitor, recover and audit
 
+A suspect budget-meter response is confirmed with at most three reads requesting
+cache revalidation inside one 15-second read window. Each request uses only the
+remaining time. Counters must still be finite, nonnegative and no lower than the last
+accepted value. `monitor.json` and the unit log retain rejected observations,
+the previous value, errors and any successful confirmation. Persistent meter
+failure stops the run; a known exhausted budget never waits for another poll.
+
+Cleanup rechecks the exact run label after removing containers and networks.
+A competing cleanup's already-removed response succeeds only when that fresh
+query proves the label empty. Unknown state or surviving resources remains a
+failure and keeps the campaign's unsettled custody.
+
 `monitor.json` records progress, key-meter spending, disk headroom and stop
 reasons. `run_manifest.json` records the seed, benchmark, immutable image ID,
 selected IDs, recovery ancestry and applied configuration; `result_index.jsonl` retains every selected task,
 including failures and tasks not started. Task artifacts live below
 `bench/dumps/`, with sanitized runtime logs in each task's `ouroboros/` folder.
-A launcher exit code is not a task score.
+A launcher exit code is not a task score. Task polling checkpoints the selected
+scrubbed logs before its status request, so an aborted run can retain partial
+evidence. HTTP delays affect that cadence; it is not a fixed two-second promise.
+Raw request/response blobs are not exported, so these copies do not constitute
+complete wire replay.
 
 Send SIGINT or SIGTERM to the launcher to stop its runner and clean up resources
 with that run's exact Docker label. Never use broad container-name cleanup on a

devtools/benchmarks/cowork_bench/campaign.py

@@ -11,24 +11,79 @@ from __future__ import annotations
 
 import json
 import math
+import os
+import subprocess
+import sys
 import pathlib
 import time
 import urllib.request
 from contextlib import contextmanager
 from typing import Iterator
 
-from devtools.benchmarks.common.manifests import write_json
+_KEY_USAGE_URL = "https://openrouter.ai/api/v1/key"
 
 
-def key_usage(api_key: str) -> float:
+class UsageCounterError(ValueError):
+    """A rejected numeric observation, retained without making it budget truth."""
+
+    def __init__(self, observed_usage: float, previous_usage: float) -> None:
+        self.observed_usage = observed_usage
+        super().__init__(f"provider usage counter decreased or is invalid: "
+                         f"observed={observed_usage!r}, previous={previous_usage!r}")
+
+
+def validate_usage(usage: float, previous_usage: float) -> None:
+    if not math.isfinite(usage) or usage < 0 or usage < previous_usage:
+        raise UsageCounterError(usage, previous_usage)
+
+
+def _read_usage_http(api_key: str, url: str, *, timeout: float) -> float:
+    """The single HTTP reader; production runs it only in the disposable worker."""
     request = urllib.request.Request(
-        "https://openrouter.ai/api/v1/key", headers={"Authorization": f"Bearer {api_key}"}
+        url, headers={"Authorization": f"Bearer {api_key}", "Cache-Control": "no-cache"},
     )
-    with urllib.request.urlopen(request, timeout=15) as response:
+    with urllib.request.urlopen(request, timeout=timeout) as response:
         payload = json.loads(response.read().decode("utf-8"))
-    value = float(payload["data"]["usage"])
-    if not math.isfinite(value) or value < 0:
-        raise ValueError("provider returned an invalid usage counter")
+    return float(payload["data"]["usage"])
+
+
+def _meter_worker() -> int:
+    payload = json.load(sys.stdin)
+    try:
+        usage = _read_usage_http(payload["api_key"], payload["url"], timeout=payload["timeout"])
+    except Exception as exc:
+        # The key travels only through stdin, never the command or diagnostic text.
+        message = str(exc).replace(payload["api_key"], "[redacted]")
+        print(json.dumps({"error_type": type(exc).__name__, "error": message}))
+        return 1
+    print(json.dumps({"usage": usage}))
+    return 0
+
+
+def key_usage(api_key: str, *, timeout: float = 15) -> float:
+    """Bound the whole HTTP read, including headers/body/DNS and worker startup."""
+    deadline = time.monotonic() + timeout
+    command = [sys.executable, "-I", "-S", str(pathlib.Path(__file__).resolve()), "--read-usage"]
+    # Preserve the parent's isolated roots, but keep this credential on stdin only.
+    env = {name: value for name, value in os.environ.items() if value != api_key}
+    payload = json.dumps({"api_key": api_key, "url": _KEY_USAGE_URL, "timeout": timeout})
+    with subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                          stderr=subprocess.PIPE, text=True, env=env) as worker:
+        try:
+            stdout, stderr = worker.communicate(payload, timeout=max(0.0, deadline - time.monotonic()))
+        except subprocess.TimeoutExpired:
+            worker.kill()
+            worker.communicate()  # Reap before returning; there is no orphan HTTP read.
+            raise TimeoutError("meter HTTP read exceeded its wall-clock deadline") from None
+        if time.monotonic() > deadline:
+            raise TimeoutError("meter HTTP response arrived after its deadline")
+        result = json.loads(stdout) if stdout else {}
+        if worker.returncode:
+            error = result.get("error") or stderr or "no worker diagnostic"
+            raise RuntimeError(f"meter HTTP read failed ({result.get('error_type', worker.returncode)}): "
+                               + str(error).replace(api_key, "[redacted]"))
+    value = float(result["usage"])
+    validate_usage(value, 0.0)
     return value
 
 
@@ -68,13 +123,14 @@ class CampaignBudget:
         return self.record["ceiling_usd"] - self.spent
 
     def save(self) -> None:
+        from devtools.benchmarks.common.manifests import write_json
+
         self.record.update({"spent_usd": self.spent, "remaining_usd": self.remaining,
                             "observed_at": time.time()})
         write_json(self.path, self.record)
 
     def observe(self, usage: float) -> None:
-        if not math.isfinite(usage) or usage < self.record["last_usage"]:
-            raise ValueError("provider usage counter decreased or is invalid; spending is unknown")
+        validate_usage(usage, self.record["last_usage"])
         self.record["last_usage"] = usage
         self.save()
 
@@ -103,3 +159,9 @@ def campaign_lock(path: pathlib.Path) -> Iterator[None]:
         yield
     finally:
         release_exclusive_file_lock(lock_path, fd)
+
+
+if __name__ == "__main__":
+    if sys.argv[1:] != ["--read-usage"]:
+        raise SystemExit("campaign.py is an internal meter worker")
+    raise SystemExit(_meter_worker())

devtools/benchmarks/cowork_bench/container/main_ouroboros.py

@@ -24,6 +24,7 @@ import pathlib
 import shutil
 import signal
 import subprocess
+import sys
 import time
 import urllib.error
 import urllib.parse
@@ -385,7 +386,9 @@ def collect_audit_artifacts(task_root: pathlib.Path, task_id: str, secrets: list
     """Copy the scrubbed Ouroboros logs next to the benchmark artefacts.
 
     The data root itself stays outside ``dumps/`` because it holds ``settings.json`` with the
-    provider credential; only value-scrubbed text leaves the container.
+    provider credential; only value-scrubbed text leaves the container. Polling
+    checkpoints publish each file atomically, so an abrupt container removal
+    leaves the previous snapshot intact even if final export never runs.
     """
     target = task_root / "ouroboros"
     target.mkdir(parents=True, exist_ok=True)
@@ -397,7 +400,9 @@ def collect_audit_artifacts(task_root: pathlib.Path, task_id: str, secrets: list
             text = source.read_text(encoding="utf-8", errors="replace")
         except OSError:
             continue
-        (target / source.name).write_text(scrub(text, secrets), encoding="utf-8")
+        staged = target / f".{source.name}.tmp"
+        staged.write_text(scrub(text, secrets), encoding="utf-8")
+        staged.replace(target / source.name)
 
 
 def write_bench_logs(task_config, status: str, started: datetime, summary: Mapping[str, Any], final_answer: str) -> None:
@@ -575,6 +580,14 @@ def run_agent_phase(task_dir: str, max_steps: int) -> str:
                 break
             if now >= outer_deadline:
                 raise WallClockInterrupt()
+            # dumps/ is host-bound; preserve available usage/tool evidence before
+            # a host stop can remove this container without running our finally.
+            try:
+                collect_audit_artifacts(task_root, task_id, secrets)
+            except OSError as exc:
+                detail = scrub(f"{type(exc).__name__}: {exc}", secrets)
+                print(f"[{ENGINE}] audit checkpoint unavailable; continuing task: {detail}",
+                      file=sys.stderr, flush=True)
             try:
                 result = _http_json("GET", f"{base_url}/api/tasks/{urllib.parse.quote(task_id)}",
                                     timeout=min(30, outer_deadline - now))

devtools/benchmarks/cowork_bench/run_cowork_bench.py

@@ -15,6 +15,7 @@ import argparse
 import hashlib
 import csv
 import json
+import math
 import os
 import pathlib
 import shutil
@@ -43,7 +44,7 @@ from devtools.benchmarks.common.result_index import (
 )
 from devtools.benchmarks.common.run_roots import assert_outside_repo, repo_root_from_devtools, run_root, timestamp_run_id
 from devtools.benchmarks.common.secrets import credential_fingerprint
-from devtools.benchmarks.cowork_bench.campaign import CampaignBudget, campaign_lock, key_usage
+from devtools.benchmarks.cowork_bench.campaign import CampaignBudget, campaign_lock, key_usage, validate_usage
 from devtools.benchmarks.cowork_bench.resource_limits import LABEL_KEY, prepare_resource_env
 from ouroboros.platform_layer import kill_process_group_id, terminate_process_group_id
 from ouroboros.process_custody import spawn_supervised
@@ -364,20 +365,39 @@ def select_tasks(bench_dir: pathlib.Path, args: argparse.Namespace, config: dict
     return remaining
 
 def remove_run_containers(docker_host: str, run_label: str) -> None:
-    """Clean only resources labeled by this invocation, including anonymous helper containers."""
+    """Settle only this invocation's resources, including concurrent cleanup removals."""
     selector = f"label={LABEL_KEY}={run_label}"
-    containers = _docker(docker_host, "ps", "-aq", "--filter", selector, timeout=60)
-    if containers.returncode != 0:
-        raise RuntimeError("cannot verify this run's container custody")
-    ids = containers.stdout.split()
-    if ids and _docker(docker_host, "rm", "-fv", *ids, timeout=120).returncode:
-        raise RuntimeError("could not stop all containers owned by this run")
-    networks = _docker(docker_host, "network", "ls", "-q", "--filter", selector, timeout=60)
-    if networks.returncode != 0:
-        raise RuntimeError("cannot verify this run's network custody")
-    ids = networks.stdout.split()
-    if ids and _docker(docker_host, "network", "rm", *ids, timeout=120).returncode:
-        raise RuntimeError("could not remove all networks owned by this run")
+    for kind, listing, removal in (
+        ("container", ("ps", "-aq"), ("rm", "-fv")),
+        ("network", ("network", "ls", "-q"), ("network", "rm")),
+    ):
+        found = _docker(docker_host, *listing, "--filter", selector, timeout=60)
+        if found.returncode:
+            raise RuntimeError(f"cannot verify this run's {kind} custody")
+        ids = found.stdout.split()
+        if not ids:
+            continue
+        removal_notice = {}
+        try:
+            removed = _docker(docker_host, *removal, *ids, timeout=120)
+            if removed.returncode or removed.stderr:
+                removal_notice = {"returncode": removed.returncode, "stdout": removed.stdout or "",
+                                  "stderr": removed.stderr or "", "exception": None}
+        except (OSError, subprocess.SubprocessError) as exc:
+            # The daemon may have completed removal despite a client failure.
+            removal_notice = {"returncode": None, "stdout": str(getattr(exc, "stdout", "") or ""),
+                              "stderr": str(getattr(exc, "stderr", "") or ""),
+                              "exception": {"type": type(exc).__name__, "message": str(exc)}}
+        if removal_notice:
+            print(json.dumps({"event": "cowork_cleanup_remove", "resource": kind,
+                              "run_label": run_label, "selected_ids": ids, **removal_notice}), flush=True)
+        remaining = _docker(docker_host, *listing, "--filter", selector, timeout=60)
+        if remaining.returncode:
+            raise RuntimeError(f"cannot verify this run's {kind} custody after removal")
+        if remaining.stdout.split():
+            print(json.dumps({"event": "cowork_cleanup_remaining", "resource": kind,
+                              "run_label": run_label, "remaining_ids": remaining.stdout.split()}), flush=True)
+            raise RuntimeError(f"could not remove all {kind}s owned by this run")
 
 
 def mark_stop(path: pathlib.Path, reason: str) -> None:
@@ -400,6 +420,51 @@ def stop_process_group(proc: subprocess.Popen) -> None:
     proc.wait(timeout=15)
 
 
+def observe_campaign_usage(api_key: str, campaign: CampaignBudget,
+                           diagnostics: list[dict[str, Any]], *, phase: str) -> None:
+    """Confirm suspect meter reads within one 15-second window, never accept a lower counter."""
+    deadline = time.monotonic() + 15.0
+    previous = campaign.record["last_usage"]
+    last_error: Exception = TimeoutError("usage confirmation window expired")
+    rejected = False
+    for attempt in range(1, 4):
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            break
+        value = None
+        try:
+            value = key_usage(api_key, timeout=remaining)
+            if time.monotonic() > deadline:
+                raise TimeoutError("usage confirmation arrived after its shared deadline")
+            validate_usage(value, previous)
+        except Exception as exc:  # Provider/transport/parse failures share the bounded confirmation.
+            last_error = exc
+            value = getattr(exc, "observed_usage", value)
+            observation = {
+                "observed_at": time.time(), "phase": phase, "attempt": attempt,
+                "previous_usage": previous,
+                "observed_usage": value if value is None or math.isfinite(value) else repr(value),
+                "accepted": False, "error_type": type(exc).__name__, "error": str(exc),
+            }
+            diagnostics.append(observation)
+            print(json.dumps({"event": "cowork_meter_observation", **observation}), flush=True)
+            rejected = True
+        else:
+            # Persistence failures are not suspect provider reads and are never retried here.
+            campaign.observe(value)
+            if rejected:
+                observation = {"observed_at": time.time(), "phase": phase, "attempt": attempt,
+                               "previous_usage": previous, "observed_usage": value, "accepted": True}
+                diagnostics.append(observation)
+                print(json.dumps({"event": "cowork_meter_observation", **observation}), flush=True)
+            return
+        if attempt < 3:
+            delay = min(3.0, max(0.0, deadline - time.monotonic()))
+            if delay:
+                time.sleep(delay)
+    raise last_error
+
+
 def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathlib.Path,
                   run_env: dict[str, str], api_key: str, campaign: CampaignBudget) -> dict[str, Any]:
     """Own runner lifetime, budget meter and disk reserve until every owned resource is settled.
@@ -410,7 +475,9 @@ def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathl
     """
     root = bench_dir.parent
     stop_file = pathlib.Path(run_env["COWORK_STOP_FILE"])
-    reserve = max(args.budget_reserve_usd, args.concurrency * args.per_task_cost_usd)
+    # A task's lifetime budget is not a reservation of unsettled provider charges.
+    # The CLI validates this explicit billing allowance as nonnegative.
+    reserve = args.budget_reserve_usd
     initial_spent = campaign.spent
     if campaign.remaining <= reserve:
         raise ValueError("campaign remaining budget does not cover the in-flight reserve")
@@ -418,6 +485,7 @@ def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathl
     proc = None
     reason = ""
     meter_error = ""
+    meter_diagnostics: list[dict[str, Any]] = []
     code = 1
     campaign.start(root)
 
@@ -436,7 +504,7 @@ def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathl
                     reason = stop_file.read_text(encoding="utf-8").strip()
                     break
                 try:
-                    campaign.observe(key_usage(api_key))
+                    observe_campaign_usage(api_key, campaign, meter_diagnostics, phase="poll")
                 except Exception as exc:
                     meter_error = type(exc).__name__
                     reason = "budget_meter_unavailable"
@@ -455,6 +523,7 @@ def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathl
                     "run_spent_usd": campaign.spent - initial_spent, "inflight_reserve_usd": reserve,
                     "disk_free_bytes": free, "root_free_bytes": root_free,
                     "ledger_counts": counts, "stop_reason": reason, "meter_error": meter_error,
+                    "meter_diagnostics": meter_diagnostics,
                 })
                 if reason:
                     mark_stop(stop_file, reason)
@@ -477,7 +546,7 @@ def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathl
             for sig, handler in old_handlers.items():
                 signal.signal(sig, handler)
         try:
-            campaign.observe(key_usage(api_key))
+            observe_campaign_usage(api_key, campaign, meter_diagnostics, phase="final")
         except Exception as exc:
             meter_error = type(exc).__name__
         campaign.finish(root, outcome=reason or "runner_finished", meter_error=meter_error)
@@ -485,10 +554,11 @@ def supervise_run(args: argparse.Namespace, command: list[str], bench_dir: pathl
             "observed_at": time.time(), "finished": True, "runner_exit_code": code,
             "campaign_spent_usd": campaign.spent, "campaign_remaining_usd": campaign.remaining,
             "run_spent_usd": campaign.spent - initial_spent, "stop_reason": reason,
-            "meter_error": meter_error,
+            "meter_error": meter_error, "meter_diagnostics": meter_diagnostics,
             "ledger_counts": write_ledger(root / "result_index.jsonl", bench_dir, args.model, args.selected_tasks),
         })
     return {"stop_reason": reason, "runner_exit_code": code, "meter_error": meter_error,
+            "meter_diagnostics": meter_diagnostics,
             "key_spend_usd": campaign.spent - initial_spent, "campaign_spent_usd": campaign.spent,
             "campaign_remaining_usd": campaign.remaining, "inflight_reserve_usd": reserve}
 

tests/test_cowork_bench_container.py

@@ -84,8 +84,7 @@ def test_container_cost_mirror_keeps_canonical_amounts_and_every_openness_field(
     assert "cost_usd" not in adapter.COST_RESULT_FIELDS
 
 
-@pytest.fixture
-def agent_episode(tmp_path, monkeypatch):
+def _make_agent_episode(tmp_path, monkeypatch):
     """Run the real entrypoint lifecycle with only process/HTTP transport replaced."""
     from types import SimpleNamespace
 
@@ -146,6 +145,11 @@ def agent_episode(tmp_path, monkeypatch):
     return state
 
 
+@pytest.fixture
+def agent_episode(tmp_path, monkeypatch):
+    return _make_agent_episode(tmp_path, monkeypatch)
+
+
 @pytest.mark.serial
 def test_setup_timeout_is_infrastructure_without_a_solve_attempt(agent_episode):
     def interrupt():
@@ -279,3 +283,119 @@ def test_real_rotation_and_export_keep_early_cost_and_tool_evidence(agent_episod
     assert audit["manual_review"] == [{"source": "tools.jsonl", "line": 1,
                                       "reason": "answer_source_or_evaluator_reference"}]
     assert not (pathlib.Path(adapter.OUROBOROS_DATA) / "archive").exists()
+
+
+@pytest.mark.serial
+def test_poll_checkpoint_survives_abrupt_exit_without_terminal_export(tmp_path):
+    import os
+    import shutil
+    import subprocess
+
+    # Reuse the real-loop fixture in a child so os._exit really skips finally.
+    # All four runtime roots are explicit; no server, Docker or model is started.
+    child = r'''
+import json, os, pathlib, runpy, sys
+import pytest
+module = runpy.run_path(sys.argv[1])
+adapter = module["adapter"]
+state = module["_make_agent_episode"](pathlib.Path(sys.argv[2]), pytest.MonkeyPatch())
+secret = "fixture-secret-must-not-leave-runtime"
+adapter.load_json = lambda path: {"settings": {"OPENROUTER_API_KEY": secret}}
+names = ("events.jsonl", "tools.jsonl", "progress.jsonl")
+def record(round_number):
+    for name in names:
+        path = pathlib.Path(adapter.OUROBOROS_DATA) / "logs" / name
+        with path.open("a", encoding="utf-8") as stream:
+            stream.write(json.dumps({"type": "llm_usage" if name == "events.jsonl" else "tool_call",
+                                     "round": round_number, "cost": 1.25, "text": secret}) + "\n")
+state.server_started = lambda: record(1)
+reads = 0
+def http(method, url, *args, **kwargs):
+    global reads
+    if method == "POST":
+        return {"task_id": "fixture-task"}
+    reads += 1
+    if reads == 1:
+        record(2)
+        return {"status": "running", "prompt_tokens": 10}
+    os._exit(73)
+adapter._http_json = http
+state.run()
+raise AssertionError("the fixture should exit before terminal export")
+'''
+    app = tmp_path / "isolated-app"
+    env = {**os.environ, "OUROBOROS_APP_ROOT": str(app),
+           "OUROBOROS_REPO_DIR": str(pathlib.Path(__file__).resolve().parents[1]),
+           "OUROBOROS_DATA_DIR": str(app / "data"),
+           "OUROBOROS_SETTINGS_PATH": str(app / "data/settings.json")}
+    peer = tmp_path / "other-task"
+    peer.mkdir()
+    (peer / "events.jsonl").write_text("unchanged\n", encoding="utf-8")
+    proc = subprocess.run([sys.executable, "-c", child, str(pathlib.Path(__file__).resolve()), str(tmp_path)],
+                          cwd=pathlib.Path(__file__).resolve().parents[1], env=env,
+                          capture_output=True, text=True, timeout=20)
+    assert proc.returncode == 73, proc.stderr
+    # Simulate removal of the private container filesystem after its hard exit.
+    shutil.rmtree(tmp_path / "data")
+    shutil.rmtree(tmp_path / "runtime")
+    dump = tmp_path / "dump"
+    assert not (dump / "ouroboros_summary.json").exists()
+    assert not (dump / "traj_log.json").exists()
+    for name in ("events.jsonl", "tools.jsonl", "progress.jsonl"):
+        text = (dump / "ouroboros" / name).read_text(encoding="utf-8")
+        assert [row["round"] for row in map(json.loads, text.splitlines())] == [1, 2]
+        assert "fixture-secret-must-not-leave-runtime" not in text
+        assert "***REDACTED***" in text
+    assert not list(dump.rglob("settings.json"))
+    assert (peer / "events.jsonl").read_text(encoding="utf-8") == "unchanged\n"
+
+
+@pytest.mark.serial
+def test_interrupted_checkpoint_publish_keeps_last_host_snapshot(tmp_path, monkeypatch):
+    data = tmp_path / "private-runtime"
+    (data / "logs").mkdir(parents=True)
+    source = data / "logs/events.jsonl"
+    source.write_text('{"round":1}\n', encoding="utf-8")
+    monkeypatch.setattr(adapter, "OUROBOROS_DATA", str(data))
+    dump = tmp_path / "host-dump"
+    adapter.collect_audit_artifacts(dump, "", [])
+    original = (dump / "ouroboros/events.jsonl").read_bytes()
+    source.write_text('{"round":1}\n{"round":2}\n', encoding="utf-8")
+    def interrupted_publish(_source, _target):
+        raise OSError("interrupted before atomic publication")
+    monkeypatch.setattr(pathlib.Path, "replace", interrupted_publish)
+    with pytest.raises(OSError, match="atomic publication"):
+        adapter.collect_audit_artifacts(dump, "", [])
+    assert (dump / "ouroboros/events.jsonl").read_bytes() == original
+
+
+@pytest.mark.serial
+def test_poll_checkpoint_write_failure_does_not_abort_task_or_final_export(agent_episode, monkeypatch, capsys):
+    secret = "checkpoint-fixture-secret"
+    monkeypatch.setattr(adapter, "load_json", lambda path: {"settings": {"OPENROUTER_API_KEY": secret}})
+    def initial_log():
+        path = pathlib.Path(adapter.OUROBOROS_DATA) / "logs/events.jsonl"
+        path.write_text(json.dumps({"type": "llm_usage", "prompt_tokens": 10, "text": secret}) + "\n",
+                        encoding="utf-8")
+    agent_episode.server_started = initial_log
+    agent_episode.results[:] = [{"status": "running"}, {"status": "completed", "final_answer": "done"}]
+    write = pathlib.Path.write_text
+    writes = []
+    def flaky_write(path, *args, **kwargs):
+        if path == agent_episode.dump / "ouroboros/.events.jsonl.tmp":
+            writes.append(path)
+            if len(writes) == 1:
+                raise OSError(f"temporary export failure involving {secret}")
+        return write(path, *args, **kwargs)
+    monkeypatch.setattr(pathlib.Path, "write_text", flaky_write)
+    summary = agent_episode.run()
+    assert summary["bench_status"] == "success"
+    assert agent_episode.calls == ["POST", "GET", "GET", "terminate", "terminate"]
+    assert len(writes) == 3  # failed checkpoint, successful next poll, final export
+    captured = capsys.readouterr()
+    assert "audit checkpoint unavailable; continuing task" in captured.err
+    assert "***REDACTED***" in captured.err
+    assert secret not in captured.err
+    exported = (agent_episode.dump / "ouroboros/events.jsonl").read_text(encoding="utf-8")
+    assert json.loads(exported)["prompt_tokens"] == 10
+    assert secret not in exported

tests/test_cowork_bench_launcher.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import io
 import os
 import pathlib
 import shutil
@@ -261,7 +262,9 @@ def supervised(tmp_path, monkeypatch):
     monkeypatch.setattr(launcher, "spawn_supervised", launch)
     monkeypatch.setattr(launcher, "stop_process_group", stop)
     monkeypatch.setattr(launcher, "remove_run_containers", cleanup)
-    monkeypatch.setattr(launcher, "key_usage", lambda _key: 100)
+    monkeypatch.setattr(launcher, "key_usage", lambda _key, **_kwargs: 100)
+    monkeypatch.setattr(launcher, "time", SimpleNamespace(
+        time=time.time, monotonic=time.monotonic, sleep=lambda _delay: None))
     monkeypatch.setattr(launcher.shutil, "disk_usage", lambda _path: SimpleNamespace(free=1024**4))
     return args, bench, env, budget, events, handlers, proc
 
@@ -273,14 +276,14 @@ def supervised(tmp_path, monkeypatch):
 def test_supervisor_stops_owned_work_on_real_boundaries(supervised, monkeypatch, trigger, expected):
     args, bench, env, budget, events, handlers, _proc = supervised
     if trigger == "meter-loss":
-        def offline(_key):
+        def offline(_key, **_kwargs):
             raise OSError("meter unavailable")
         monkeypatch.setattr(launcher, "key_usage", offline)
     elif trigger == "disk":
         args.min_free_gib = 200
         monkeypatch.setattr(launcher.shutil, "disk_usage", lambda _path: SimpleNamespace(free=199 * 1024**3))
     else:
-        monkeypatch.setattr(launcher, "key_usage", lambda _key: 1000 if trigger == "campaign" else 250)
+        monkeypatch.setattr(launcher, "key_usage", lambda _key, **_kwargs: 1000 if trigger == "campaign" else 250)
     result = launcher.supervise_run(args, ["fake-runner"], bench, env, "not-a-real-key", budget)
     assert result["stop_reason"] == expected
     assert events == ["stop-group", "cleanup-owned"]
@@ -292,6 +295,40 @@ def test_supervisor_stops_owned_work_on_real_boundaries(supervised, monkeypatch,
         assert result["meter_error"] == "OSError"
 
 
+@pytest.mark.parametrize("reserve", [0.0, 100.0, 250.0])
+def test_parallel_campaign_spends_past_lifetime_caps_then_stops_at_explicit_reserve(supervised, monkeypatch, reserve):
+    args, bench, env, _old_budget, events, _handlers, proc = supervised
+    args.concurrency = 32
+    args.per_task_cost_usd = 25
+    args.budget_usd = 2000
+    args.budget_reserve_usd = reserve
+    budget = budgets.CampaignBudget(bench.parent / "full-campaign.json", fingerprint="key-a",
+                                   ceiling=2000, usage=100)
+    # Meter values are cumulative key usage. Continue after $1200 (the old
+    # 32*$25 reserve stopped here) and $1600; stop only at the selected margin.
+    usage = iter([1300, 1700, 2100 - reserve, 2100 - reserve])
+    monkeypatch.setattr(launcher, "key_usage", lambda _key, **_kwargs: next(usage))
+    continued_at = []
+    def still_running(**kwargs):
+        continued_at.append(budget.spent)
+        raise subprocess.TimeoutExpired("fake-runner", kwargs["timeout"])
+    proc.wait = still_running
+    result = launcher.supervise_run(args, ["fake-runner"], bench, env, "not-a-real-key", budget)
+    assert continued_at == [1200, 1600]
+    assert result["stop_reason"] == "campaign_budget_reserve"
+    assert result["campaign_spent_usd"] == 2000 - reserve
+    assert result["campaign_remaining_usd"] == reserve
+    assert result["inflight_reserve_usd"] == reserve
+    assert events == ["stop-group", "cleanup-owned"]
+    assert "active_run" not in budget.record
+
+
+def test_negative_explicit_billing_reserve_is_rejected():
+    with pytest.raises(SystemExit) as exc:
+        launcher.parse_args(["--budget-reserve-usd", "-1"])
+    assert exc.value.code == 2
+
+
 def test_supervisor_handles_sigterm_then_cleans_only_owned_resources(supervised, monkeypatch):
     args, bench, env, budget, events, handlers, proc = supervised
     def launch(*_args, **kwargs):
@@ -323,15 +360,20 @@ def test_cleanup_failure_keeps_campaign_custody_unsettled(supervised, monkeypatc
 
 def test_cleanup_selection_cannot_include_a_peer_run(monkeypatch):
     calls = []
+    remaining = {"containers": "own-a\nown-b\n", "networks": "own-net\n"}
     def docker(host, *argv, **_kwargs):
         assert host == "unix:///owned.sock"
         calls.append(argv)
         if argv[:2] == ("ps", "-aq"):
             assert argv[-1] == f"label={launcher.LABEL_KEY}=owned-run"
-            return subprocess.CompletedProcess(argv, 0, "own-a\nown-b\n")
+            return subprocess.CompletedProcess(argv, 0, remaining["containers"])
         if argv[:3] == ("network", "ls", "-q"):
             assert argv[-1] == f"label={launcher.LABEL_KEY}=owned-run"
-            return subprocess.CompletedProcess(argv, 0, "own-net\n")
+            return subprocess.CompletedProcess(argv, 0, remaining["networks"])
+        if argv[:2] == ("rm", "-fv"):
+            remaining["containers"] = ""
+        if argv[:2] == ("network", "rm"):
+            remaining["networks"] = ""
         return subprocess.CompletedProcess(argv, 0, "")
     monkeypatch.setattr(launcher, "_docker", docker)
     launcher.remove_run_containers("unix:///owned.sock", "owned-run")
@@ -573,8 +615,8 @@ def test_paid_runner_uses_immutable_image_and_scrubs_ambient_alternate_keys(dry_
     monkeypatch.setenv("OPENROUTER_API_KEY", "test-selected-key")
     monkeypatch.setenv("LLM_API_KEY", "test-unrelated-llm-key")
     monkeypatch.setenv("MODEL_API_KEY", "test-unrelated-model-key")
-    monkeypatch.setattr(launcher, "key_headroom", lambda _key: {"effective": 1000})
-    monkeypatch.setattr(launcher, "key_usage", lambda _key: 100)
+    monkeypatch.setattr(launcher, "key_headroom", lambda _key, **_kwargs: {"effective": 1000})
+    monkeypatch.setattr(launcher, "key_usage", lambda _key, **_kwargs: 100)
     monkeypatch.setattr(launcher, "prepare_resource_env", lambda env, **_kwargs: dict(env))
     def supervise(args, command, bench, env, api_key, campaign):
         assert command[-1] == "one"
@@ -592,3 +634,231 @@ def test_paid_runner_uses_immutable_image_and_scrubs_ambient_alternate_keys(dry_
     assert launcher.main([*paid_argv, "--campaign-file", str(out.parent / "campaign.json")]) == 0
     assert (out / "bench" / "configs" / launcher.SECRET_NAME).read_text(encoding="utf-8") == "{}"
     assert "test-selected-key" not in (out / "run_manifest.json").read_text(encoding="utf-8")
+
+
+def test_meter_http_read_requests_uncached_data_with_remaining_timeout(monkeypatch):
+    received = []
+    def response(request, *, timeout):
+        received.append((dict(request.header_items()), timeout))
+        return io.BytesIO(b'{"data":{"usage":123.5}}')
+    monkeypatch.setattr(budgets.urllib.request, "urlopen", response)
+    assert budgets._read_usage_http("not-a-real-key", budgets._KEY_USAGE_URL, timeout=2.25) == 123.5
+    assert received[0][0]["Cache-control"] == "no-cache"
+    assert received[0][1] == 2.25
+
+
+def test_backward_meter_read_is_confirmed_without_stopping_the_run(supervised, monkeypatch, capsys):
+    args, bench, env, budget, events, _handlers, proc = supervised
+    observations = iter([99.0, 101.0, 102.0])
+    monkeypatch.setattr(launcher, "key_usage", lambda _key, **_kwargs: next(observations))
+    def completed(**_kwargs):
+        proc.returncode = 0
+        return 0
+    proc.wait = completed
+    result = launcher.supervise_run(args, ["fake-runner"], bench, env, "not-a-real-key", budget)
+    assert result["stop_reason"] == ""
+    assert result["meter_error"] == ""
+    assert budget.record["last_usage"] == 102.0
+    assert budget.spent == 2.0
+    diagnostics = result["meter_diagnostics"]
+    assert [(row["observed_usage"], row["accepted"]) for row in diagnostics] == [(99.0, False), (101.0, True)]
+    assert diagnostics[0]["previous_usage"] == 100
+    final = json.loads((bench.parent / "monitor.json").read_text(encoding="utf-8"))
+    assert final["finished"] is True and final["meter_diagnostics"] == diagnostics
+    assert '"observed_usage": 99.0' in capsys.readouterr().out
+    assert events == ["stop-group", "cleanup-owned"]
+
+
+@pytest.mark.parametrize("invalid", [-1.0, float("nan"), float("inf")])
+def test_confirmation_never_accepts_an_invalid_numeric_usage(supervised, monkeypatch, invalid):
+    _args, _bench, _env, budget, _events, _handlers, _proc = supervised
+    values = iter([invalid, 101.0])
+    monkeypatch.setattr(launcher, "key_usage", lambda _key, **_kwargs: next(values))
+    diagnostics = []
+    launcher.observe_campaign_usage("not-a-real-key", budget, diagnostics, phase="poll")
+    assert budget.record["last_usage"] == 101
+    assert diagnostics[0]["accepted"] is False
+    assert diagnostics[0]["previous_usage"] == 100
+    # Non-finite rejected observations remain printable JSON evidence, never NaN budget truth.
+    json.dumps(diagnostics, allow_nan=False)
+
+
+def test_persistent_bad_counter_stops_and_keeps_rejected_values_in_final_monitor(supervised, monkeypatch):
+    args, bench, env, budget, _events, _handlers, _proc = supervised
+    calls = []
+    def stale(_key, *, timeout):
+        calls.append(timeout)
+        return 99.0
+    monkeypatch.setattr(launcher, "key_usage", stale)
+    result = launcher.supervise_run(args, ["fake-runner"], bench, env, "not-a-real-key", budget)
+    assert result["stop_reason"] == "budget_meter_unavailable"
+    assert budget.record["last_usage"] == 100
+    assert len(calls) == 6  # One bounded confirmation in the loop and one final settlement read.
+    assert {row["phase"] for row in result["meter_diagnostics"]} == {"poll", "final"}
+    assert all(row["observed_usage"] == 99.0 and not row["accepted"] for row in result["meter_diagnostics"])
+    final = json.loads((bench.parent / "monitor.json").read_text(encoding="utf-8"))
+    assert final["meter_diagnostics"] == result["meter_diagnostics"]
+    assert final["meter_error"] == "UsageCounterError"
+
+
+def test_confirmation_shares_one_timeout_window_instead_of_three_full_timeouts(supervised, monkeypatch):
+    _args, _bench, _env, budget, _events, _handlers, _proc = supervised
+    clock = SimpleNamespace(now=0.0)
+    def advance(delay):
+        clock.now += delay
+    monkeypatch.setattr(launcher, "time", SimpleNamespace(
+        monotonic=lambda: clock.now, time=lambda: clock.now, sleep=advance))
+    timeouts = []
+    def unavailable(_key, *, timeout):
+        timeouts.append(timeout)
+        advance(min(7.0, timeout))
+        raise OSError("meter unavailable")
+    monkeypatch.setattr(launcher, "key_usage", unavailable)
+    original = budget.path.read_bytes()
+    diagnostics = []
+    with pytest.raises(OSError, match="meter unavailable"):
+        launcher.observe_campaign_usage("not-a-real-key", budget, diagnostics, phase="poll")
+    assert timeouts == [15.0, 5.0]
+    assert clock.now == 15.0
+    assert len(diagnostics) == 2
+    assert budget.path.read_bytes() == original
+
+
+def test_confirmed_exhausted_budget_stops_without_waiting_for_another_poll(supervised, monkeypatch):
+    args, bench, env, budget, _events, _handlers, proc = supervised
+    calls = []
+    def exhausted(_key, *, timeout):
+        calls.append(timeout)
+        return 1000.0
+    monkeypatch.setattr(launcher, "key_usage", exhausted)
+    proc.wait = lambda **_kwargs: pytest.fail("known exhausted budget must not continue work")
+    result = launcher.supervise_run(args, ["fake-runner"], bench, env, "not-a-real-key", budget)
+    assert result["stop_reason"] == "campaign_budget_reserve"
+    assert len(calls) == 2  # Poll plus final settlement, with no confirmation retry.
+    assert result["meter_diagnostics"] == []
+
+
+@pytest.mark.parametrize("kind", ["container", "network"])
+def test_cleanup_racing_404_is_success_only_after_exact_label_is_empty(monkeypatch, kind):
+    seen = []
+    removed = False
+    def docker(host, *argv, **_kwargs):
+        nonlocal removed
+        assert host == "unix:///owned.sock"
+        seen.append(argv)
+        selected = argv[:2] == ("ps", "-aq") if kind == "container" else argv[:3] == ("network", "ls", "-q")
+        if selected:
+            assert argv[-1] == f"label={launcher.LABEL_KEY}=owned-run"
+            return subprocess.CompletedProcess(argv, 0, "" if removed else "own-id\n")
+        if argv[:2] in (("rm", "-fv"), ("network", "rm")):
+            removed = True  # A competing cleanup got there first.
+            return subprocess.CompletedProcess(argv, 1, "", "No such resource")
+        return subprocess.CompletedProcess(argv, 0, "")
+    monkeypatch.setattr(launcher, "_docker", docker)
+    launcher.remove_run_containers("unix:///owned.sock", "owned-run")
+    assert removed
+    assert sum(argv[-1] == f"label={launcher.LABEL_KEY}=owned-run" for argv in seen) == 3
+
+
+@pytest.mark.parametrize("remove_exit", [0, 1])
+def test_cleanup_never_claims_surviving_containers_are_gone(monkeypatch, remove_exit):
+    def docker(_host, *argv, **_kwargs):
+        if argv[:2] == ("ps", "-aq"):
+            return subprocess.CompletedProcess(argv, 0, "still-owned\n")
+        return subprocess.CompletedProcess(argv, remove_exit, "")
+    monkeypatch.setattr(launcher, "_docker", docker)
+    with pytest.raises(RuntimeError, match="could not remove all containers"):
+        launcher.remove_run_containers("unix:///owned.sock", "owned-run")
+
+
+def test_cleanup_failed_relist_is_unknown_even_when_stdout_is_empty(monkeypatch):
+    lists = 0
+    def docker(_host, *argv, **_kwargs):
+        nonlocal lists
+        if argv[:2] == ("ps", "-aq"):
+            lists += 1
+            return subprocess.CompletedProcess(argv, 0 if lists == 1 else 1, "own-id\n" if lists == 1 else "")
+        return subprocess.CompletedProcess(argv, 1, "", "No such resource")
+    monkeypatch.setattr(launcher, "_docker", docker)
+    with pytest.raises(RuntimeError, match="cannot verify.*custody after removal"):
+        launcher.remove_run_containers("unix:///owned.sock", "owned-run")
+
+
+def test_lagging_counter_gets_time_to_catch_up_without_weakening_monotonicity(supervised, monkeypatch):
+    _args, _bench, _env, budget, _events, _handlers, _proc = supervised
+    clock = SimpleNamespace(now=0.0)
+    def advance(delay):
+        clock.now += delay
+    monkeypatch.setattr(launcher, "time", SimpleNamespace(
+        monotonic=lambda: clock.now, time=lambda: clock.now, sleep=advance))
+    reads = []
+    def provider(_key, *, timeout):
+        reads.append((clock.now, timeout))
+        # Neither the stale value nor the waiting period may lower durable spending.
+        assert budget.record["last_usage"] == 100
+        assert json.loads(budget.path.read_text(encoding="utf-8"))["last_usage"] == 100
+        return 99.0 if clock.now < 3.0 else 101.0
+    monkeypatch.setattr(launcher, "key_usage", provider)
+    diagnostics = []
+    launcher.observe_campaign_usage("not-a-real-key", budget, diagnostics, phase="poll")
+    assert reads == [(0.0, 15.0), (3.0, 12.0)]
+    assert budget.record["last_usage"] == 101
+    assert [(row["observed_usage"], row["accepted"]) for row in diagnostics] == [(99.0, False), (101.0, True)]
+    assert clock.now == 3.0
+
+
+@pytest.mark.parametrize("removal", ["stderr-with-zero", "nonzero", "exception"])
+def test_cleanup_retains_actual_removal_diagnostics_without_requiring_nonzero(monkeypatch, capsys, removal):
+    listed = 0
+    def docker(_host, *argv, **_kwargs):
+        nonlocal listed
+        if argv[:2] == ("ps", "-aq"):
+            listed += 1
+            return subprocess.CompletedProcess(argv, 0, "own-id\n" if listed == 1 else "")
+        if argv[:2] == ("rm", "-fv"):
+            if removal == "exception":
+                raise OSError("daemon connection reset")
+            return subprocess.CompletedProcess(argv, int(removal == "nonzero"), "", "No such container: own-id")
+        return subprocess.CompletedProcess(argv, 0, "")
+    monkeypatch.setattr(launcher, "_docker", docker)
+    launcher.remove_run_containers("unix:///owned.sock", "owned-run")
+    record = json.loads(capsys.readouterr().out)
+    assert record["event"] == "cowork_cleanup_remove"
+    assert record["run_label"] == "owned-run"
+    assert record["selected_ids"] == ["own-id"]
+    if removal == "exception":
+        assert record["exception"] == {"type": "OSError", "message": "daemon connection reset"}
+        assert record["returncode"] is None
+    else:
+        assert record["returncode"] == int(removal == "nonzero")
+        assert record["stderr"] == "No such container: own-id"
+        assert record["stdout"] == ""
+
+
+def test_cleanup_logs_survivor_ids_and_still_refuses_custody_release(monkeypatch, capsys):
+    def docker(_host, *argv, **_kwargs):
+        return subprocess.CompletedProcess(argv, 0, "owned-survivor\n" if argv[:2] == ("ps", "-aq") else "")
+    monkeypatch.setattr(launcher, "_docker", docker)
+    with pytest.raises(RuntimeError, match="could not remove all containers"):
+        launcher.remove_run_containers("unix:///owned.sock", "owned-run")
+    record = json.loads(capsys.readouterr().out)
+    assert record == {"event": "cowork_cleanup_remaining", "resource": "container",
+                      "run_label": "owned-run", "remaining_ids": ["owned-survivor"]}
+
+
+def test_late_valid_confirmation_is_not_accepted_even_if_reader_returns_it(supervised, monkeypatch):
+    _args, _bench, _env, budget, _events, _handlers, _proc = supervised
+    clock = SimpleNamespace(now=0.0)
+    monkeypatch.setattr(launcher, "time", SimpleNamespace(
+        monotonic=lambda: clock.now, time=lambda: clock.now,
+        sleep=lambda seconds: setattr(clock, "now", clock.now + seconds)))
+    def late(_key, *, timeout):
+        clock.now += timeout + 1
+        return 101.0
+    monkeypatch.setattr(launcher, "key_usage", late)
+    diagnostics = []
+    with pytest.raises(TimeoutError, match="shared deadline"):
+        launcher.observe_campaign_usage("not-a-real-key", budget, diagnostics, phase="poll")
+    assert budget.record["last_usage"] == 100
+    assert diagnostics[0]["observed_usage"] == 101
+    assert diagnostics[0]["accepted"] is False

tests/test_cowork_bench_meter_transport.py

@@ -0,0 +1,119 @@
+"""Real loopback HTTP confirms the meter's wall-clock bound without a provider."""
+from __future__ import annotations
+
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, HTTPServer
+import json
+import os
+import subprocess
+import threading
+import time
+
+import pytest
+
+from devtools.benchmarks.cowork_bench import campaign
+
+pytestmark = pytest.mark.serial
+
+
+@contextmanager
+def meter_endpoint(header_delay, body_delay, usage):
+    seen = []
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            seen.append(dict(self.headers))
+            body = json.dumps({"data": {"usage": usage}}).encode("utf-8")
+            try:
+                time.sleep(header_delay)
+                self.send_response(200)
+                self.send_header("Content-Length", str(len(body)))
+                self.end_headers()
+                time.sleep(body_delay)
+                self.wfile.write(body)
+            except (BrokenPipeError, ConnectionResetError):
+                pass  # The expired client's worker was killed and reaped.
+
+        def log_message(self, *_args):
+            pass
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=lambda: server.serve_forever(poll_interval=0.01), daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}/key", seen
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=3)
+        assert not thread.is_alive()
+
+
+@pytest.fixture
+def meter_worker(monkeypatch, tmp_path):
+    roots = {
+        "OUROBOROS_APP_ROOT": tmp_path,
+        "OUROBOROS_REPO_DIR": tmp_path / "repo",
+        "OUROBOROS_DATA_DIR": tmp_path / "data",
+        "OUROBOROS_SETTINGS_PATH": tmp_path / "data" / "settings.json",
+    }
+    # Exercise CI's absent outer environment after safe module imports. The
+    # fixture supplies the isolated roots whose worker inheritance we assert.
+    for name in roots:
+        monkeypatch.delenv(name, raising=False)
+    for name, path in roots.items():
+        monkeypatch.setenv(name, str(path))
+    key = "synthetic-meter-key-never-in-argv"
+    monkeypatch.setenv("TEST_METER_SECRET", key)
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost")
+    real_popen = subprocess.Popen
+    workers = []
+    def capture(command, **kwargs):
+        assert key not in " ".join(command)
+        assert key not in kwargs["env"].values()
+        assert command[1:3] == ["-I", "-S"]  # No editable finder or live imports.
+        for name in roots:
+            assert kwargs["env"][name] == os.environ[name]
+        worker = real_popen(command, **kwargs)
+        workers.append(worker)
+        return worker
+    monkeypatch.setattr(campaign.subprocess, "Popen", capture)
+    yield key, workers
+    assert len(workers) == 1
+    assert workers[0].poll() is not None
+    if os.name != "nt":
+        with pytest.raises(ChildProcessError):
+            os.waitpid(workers[0].pid, os.WNOHANG)
+
+
+def test_delayed_headers_and_body_complete_inside_one_deadline(monkeypatch, meter_worker):
+    key, workers = meter_worker
+    with meter_endpoint(0.1, 0.15, 123.5) as (url, seen):
+        monkeypatch.setattr(campaign, "_KEY_USAGE_URL", url)
+        assert campaign.key_usage(key, timeout=1.5) == 123.5
+        assert seen[0]["Authorization"] == "Bearer " + key
+        assert seen[0]["Cache-Control"] == "no-cache"
+        assert workers[0].returncode == 0
+
+
+def test_headers_and_body_cannot_each_consume_the_full_timeout(monkeypatch, meter_worker):
+    key, workers = meter_worker
+    # Each delay is below 1.5 s; their sum is above it, reproducing the socket-timeout gap.
+    with meter_endpoint(0.9, 0.9, 123.5) as (url, _seen):
+        monkeypatch.setattr(campaign, "_KEY_USAGE_URL", url)
+        started = time.monotonic()
+        with pytest.raises(TimeoutError, match="wall-clock deadline") as caught:
+            campaign.key_usage(key, timeout=1.5)
+        elapsed = time.monotonic() - started
+        assert elapsed < 1.75
+        assert workers[0].poll() is not None
+        assert key not in str(caught.value)
+
+
+def test_meter_worker_error_never_exposes_the_stdin_credential(monkeypatch, meter_worker):
+    key, _workers = meter_worker
+    with meter_endpoint(0, 0, key) as (url, _seen):
+        monkeypatch.setattr(campaign, "_KEY_USAGE_URL", url)
+        with pytest.raises(RuntimeError) as caught:
+            campaign.key_usage(key, timeout=1.5)
+        assert key not in str(caught.value)
+        assert "[redacted]" in str(caught.value)