Close local runtime loose ends and record external blockers

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e387-534d-70e3-ad53-4ea05676db8c

scripts/prove-metered-runtime.py

@@ -42,10 +42,10 @@ BETAS = (
 DUMMY_KEY = "synthetic-provider-key-no-live-authority"
 
 
-def response_bytes():
+def response_bytes(model, tool=False):
     events = [
         {"type": "message_start", "message": {"id": "msg_fixture", "type": "message",
-         "role": "assistant", "model": "claude-sonnet-4-6", "content": [],
+         "role": "assistant", "model": model, "content": [],
          "stop_reason": None, "stop_sequence": None,
          "usage": {"input_tokens": 100, "output_tokens": 0,
                    "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}}},
@@ -59,6 +59,10 @@ def response_bytes():
                    "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}},
         {"type": "message_stop"},
     ]
+    if tool:
+        events[1]["content_block"] = {"type": "tool_use", "id": "fixture_git_status", "name": "Bash", "input": {}}
+        events[2]["delta"] = {"type": "input_json_delta", "partial_json": json.dumps({"command": "git status --short", "description": "Synthetic permitted tool proof"})}
+        events[4]["delta"]["stop_reason"] = "tool_use"
     return "".join("event: " + x["type"] + "\ndata: " + json.dumps(x) + "\n\n" for x in events).encode()
 
 
@@ -66,6 +70,9 @@ class FixtureServer(http.server.ThreadingHTTPServer):
     provider_calls = 0
     queue_claims = 0
     provider_key_correct = False
+    expected_model = ""
+    request_shapes = None
+    tool_cycle = False
 
 
 class Handler(http.server.BaseHTTPRequestHandler):
@@ -86,19 +93,43 @@ class Handler(http.server.BaseHTTPRequestHandler):
             self.send_error(404)
 
     def do_POST(self):
-        self.rfile.read(int(self.headers.get("Content-Length", "0")))
+        raw = self.rfile.read(int(self.headers.get("Content-Length", "0")))
         if self.path == "/ops-runs/claim":
             self.server.queue_claims += 1
             self.reply(b'{"items":[]}')
         elif self.path == "/v1/messages":
             self.server.provider_calls += 1
             self.server.provider_key_correct = self.headers.get("x-api-key") == DUMMY_KEY
-            self.reply(response_bytes(), "text/event-stream")
+            request = json.loads(raw)
+            thinking = request.get("thinking", {})
+            shape = {"model": request.get("model"), "max_tokens": request.get("max_tokens"),
+                     "thinking_type": thinking.get("type"),
+                     "manual_thinking_budget": "budget_tokens" in thinking,
+                     "sampling_parameters": {k: request[k] for k in ("temperature", "top_p", "top_k") if k in request},
+                     "betas": self.headers.get("anthropic-beta", "").split(","),
+                     "tool_results": [{"id": b.get("tool_use_id"), "is_error": b.get("is_error", False)}
+                                      for m in request.get("messages", [])
+                                      if isinstance(m.get("content"), list)
+                                      for b in m["content"] if b.get("type") == "tool_result"]}
+            self.server.request_shapes.append(shape)
+            if shape["model"] != self.server.expected_model:
+                self.send_error(400)
+                return
+            # Sonnet 5 rejects manual thinking and non-default sampling settings.
+            # Match that documented refusal rather than hiding it behind a 200.
+            if shape["model"] == "claude-sonnet-5" and (
+                shape["thinking_type"] == "enabled" or shape["manual_thinking_budget"]
+                or any(value != {"temperature": 1, "top_p": 1, "top_k": 0}[key]
+                       for key, value in shape["sampling_parameters"].items())
+            ):
+                self.send_error(400)
+                return
+            self.reply(response_bytes(self.server.expected_model, self.server.tool_cycle and len(self.server.request_shapes) == 1), "text/event-stream")
         else:
             self.send_error(404)
 
 
-def case(root, runtime, checksum, server, cap, bootstrap=False):
+def case(root, runtime, checksum, server, cap, args, bootstrap=False):
     root.mkdir()
     private = root / "private"; private.mkdir(mode=0o700)
     source = root / "source"; source.mkdir()
@@ -106,7 +137,9 @@ def case(root, runtime, checksum, server, cap, bootstrap=False):
     (source / "README.md").write_text("Disposable fixture. Do not edit files.\n")
     profiles = root / "profiles"; profiles.mkdir()
     catalog = ProfileCatalog()
-    profile, descriptor = catalog.resolve("harness.agent-dev-local@1.0.0")
+    profile, descriptor = catalog.resolve(args.profile_ref)
+    assert str(profile.ref) == args.profile_ref, "resolved profile mismatch"
+    assert profile.model.model == args.expected_model, "resolved model mismatch"
     profile = profile.model_copy(update={
         "limits": ExecutionLimits(max_budget_usd=cap, max_turns=4),
         "operational_readiness": OperationalReadiness(status="ready", reason="disposable fixture only",
@@ -121,15 +154,34 @@ def case(root, runtime, checksum, server, cap, bootstrap=False):
         target_repo=str(source), project="fixture-factory", profile_ref=str(profile.ref),
         profile_sha256=digest(profile.model_dump(mode="json")),
         descriptor_sha256=digest(descriptor.model_dump(mode="json")), repository_grant_id=grant.grant_id,
-        max_budget_usd=str(cap), max_liability_usd="5" if cap > 0.01 else "0.01", max_turns=4,
-        eur_per_usd="1", per_run_eur="5", daily_eur="10", total_eur="15",
+        max_budget_usd=str(cap), max_liability_usd=("10" if args.tool_cycle else "5") if cap > 0.01 else "0.01", max_turns=4,
+        eur_per_usd="1", per_run_eur="10" if args.tool_cycle else "5", daily_eur="20", total_eur="30",
     )
     parent = SpendLedger(private / "spend.sqlite3", policy); parent.initialize()
     meter = RequestLedger(parent); meter.initialize()
     policy_path = private / "spend-policy.json"
     policy_path.write_text(json.dumps(policy.__dict__)); policy_path.chmod(0o600)
-    messages = MessagesPolicy("fixture:not-live-prices", profile.model.model, 200000, 32000,
-                              3, 15, allowed_betas=BETAS)
+    admission_requests = []
+    class RecordingPolicy(MessagesPolicy):
+        def validate(self, data, betas):
+            shape = {"model": data.get("model"), "max_tokens": data.get("max_tokens"),
+                     "thinking_type": data.get("thinking", {}).get("type"),
+                     "top_level_fields": sorted(data), "betas": betas.split(","),
+                     "message_roles": [m.get("role") for m in data.get("messages", [])],
+                     "output_config_keys": sorted(data.get("output_config", {})),
+                     "output_effort": data.get("output_config", {}).get("effort")}
+            admission_requests.append(shape)
+            try:
+                return super().validate(data, betas)
+            except Exception as exc:
+                shape["refusal"] = str(exc)
+                raise
+    messages = RecordingPolicy("fixture:not-live-prices", profile.model.model,
+                              args.context_tokens, args.max_output_tokens, args.input_rate, args.output_rate,
+                              allowed_betas=BETAS + tuple(args.extra_beta))
+    server.expected_model = profile.model.model
+    server.request_shapes = []
+    server.tool_cycle = args.tool_cycle
     owner_path = private / "owner.json"
     owner_path.write_text(json.dumps({"version": "1", "authority_ref": policy.authority_ref,
         "spend_policy_sha256": policy.sha256, "messages_policy": messages.__dict__,
@@ -199,9 +251,12 @@ print(json.dumps({'runtime_readonly':readonly,'python_prefix':sys.prefix,
                 assert executed.exit_code != 0 and terminal["is_error"]
                 assert server.provider_calls == before and not meter.status()
             else:
-                assert executed.exit_code == 0 and not terminal.get("is_error")
-                assert server.provider_calls == before + 1 and server.provider_key_correct
-                assert len(meter.status()) == 1 and meter.status()[0]["state"] == "charged"
+                assert executed.exit_code == 0 and not terminal.get("is_error"), json.dumps({"exit_code": executed.exit_code, "subtype": terminal.get("subtype"), "request_shapes": server.request_shapes, "admission_requests": admission_requests})
+                assert server.provider_calls == before + (2 if args.tool_cycle else 1) and server.provider_key_correct
+                assert len(meter.status()) == (2 if args.tool_cycle else 1)
+                assert all(row["state"] == "charged" for row in meter.status())
+                if args.tool_cycle:
+                    assert {"id": "fixture_git_status", "is_error": False} in server.request_shapes[-1]["tool_results"]
             token = manager._messages_route.token
         finally:
             manager.destroy(status.sandbox_id)
@@ -212,7 +267,9 @@ print(json.dumps({'runtime_readonly':readonly,'python_prefix':sys.prefix,
               "provider_requests": server.provider_calls-before, "cli_exit_code": executed.exit_code,
               "cli_is_error": terminal.get("is_error"), "cli_estimated_usd": terminal.get("total_cost_usd"),
               "request_reservations": meter.status(), "route_revoked": True, "workspace_removed": True,
-              "bootstrap": bootstrap_result}
+              "bootstrap": bootstrap_result, "profile_ref": args.profile_ref,
+              "model": profile.model.model, "request_shapes": list(server.request_shapes),
+              "policy_context_tokens": messages.context_tokens, "admission_requests": admission_requests}
     assert token not in json.dumps(output) and DUMMY_KEY not in json.dumps(output)
     return output
 
@@ -221,6 +278,15 @@ def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--runtime", required=True, type=Path)
     parser.add_argument("--sha256", required=True)
+    parser.add_argument("--profile-ref", required=True,
+                        help="Exact bundled profile under test; never infer from the artifact name")
+    parser.add_argument("--expected-model", required=True)
+    parser.add_argument("--tool-cycle", action="store_true", help="Two fake-provider calls and an actual permitted git-status tool, synthetic EUR 10 envelope")
+    parser.add_argument("--context-tokens", type=int, default=200000)
+    parser.add_argument("--max-output-tokens", type=int, default=32000)
+    parser.add_argument("--extra-beta", action="append", default=[])
+    parser.add_argument("--input-rate", type=int, default=3)
+    parser.add_argument("--output-rate", type=int, default=15)
     args = parser.parse_args()
     runtime = args.runtime.resolve()
     assert sys.flags.isolated and sys.dont_write_bytecode, "run candidate python with -I -B"
@@ -238,13 +304,14 @@ def main():
         server = FixtureServer(("127.0.0.1", 0), Handler)
         threading.Thread(target=server.serve_forever, daemon=True).start()
         try:
-            positive = case(root / "positive", runtime, args.sha256, server, 1.0, bootstrap=True)
-            refused = case(root / "refused", runtime, args.sha256, server, 0.01)
+            positive = case(root / "positive", runtime, args.sha256, server, 1.0, args, bootstrap=True)
+            refused = case(root / "refused", runtime, args.sha256, server, 0.01, args)
         finally:
             server.shutdown(); server.server_close()
     assert runtime_digest(runtime) == args.sha256, "proof mutated the candidate artifact"
     print(json.dumps({"scope": "standalone installed packages and real CLI/bwrap; synthetic key/provider/empty queue",
         "ok": True, "runtime_sha256": args.sha256,
+        "proof_script_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
         "claude_sha256": hashlib.sha256((runtime/'bin/claude').read_bytes()).hexdigest(),
         "imports": imports, "packaged_definitions": True, "positive": positive, "refused": refused,
         "artifact_unchanged": True, "live_factory_attempts": 0}, indent=2))