diff --git a/docs/kv-live-launcher.md b/docs/kv-live-launcher.md index 1e9ab586f..401a6f8bf 100644 --- a/docs/kv-live-launcher.md +++ b/docs/kv-live-launcher.md @@ -46,6 +46,19 @@ isolation/runtime/cleanup/replay receipts. Each preparation stage records its start before running. Missing, failed or interrupted stages are explicit; no exception text, process command, environment or arbitrary log is published. +The execution bundle also projects closed worker diagnostics from fixed trusted +result paths: worker status, an allowlisted reason/error code, `provider_attempts` +and outer executor status/return code/elapsed seconds. `provider_attempts` counts +invocations of the transport function, **not confirmed HTTP delivery or charges**. +A present zero is distinct from a missing/malformed/interrupted record (`null`). +Only fixed reason enums or `http_100` through `http_599` are retained; unknown +exceptions become `unknown`, never arbitrary exception text. The transport code +is captured separately before the edit agent generalizes its failure. These +diagnostics do not settle costs, authorize retry or upgrade a verdict. Packaging +only reads existing first-run files, including when controller writing stopped. +This future retention correction cannot recover a reason omitted from an earlier +artifact; that earlier cause and billing remain unknown without separate evidence. + The grader uploads only its dedicated `public/` projection, not its working directory. `evidence.json` enumerates every assigned hidden trace and the fixed alternating AB/BA timing trials plus quality validation. It binds the original diff --git a/tests/test_kv_live_job.py b/tests/test_kv_live_job.py index 850511549..879a79bb1 100644 --- a/tests/test_kv_live_job.py +++ b/tests/test_kv_live_job.py @@ -105,7 +105,7 @@ def staged(tmp_path, monkeypatch): configs=configs, authorization=authorization) -def launcher(scripts, seen): +def launcher(scripts, seen, *, transport=None): def execute(command, cwd, output, deadline, *, environment): assert environment[provider.KEY_ENV] == "OFFLINE-FAKE-PROVIDER-KEY" forbidden = {"GITHUB_TOKEN", "ACTIONS_RUNTIME_TOKEN", "HOME", "PYTHONPATH"} @@ -117,7 +117,7 @@ def execute(command, cwd, output, deadline, *, environment): responses = iter(copy.deepcopy(scripts)) code = worker.live_worker(args["--inputs"], args["--workspace"], args["--output"], deadline, environment=environment, emit=lambda _: None, - transport=lambda *a, **kw: next(responses)) + transport=transport or (lambda *a, **kw: next(responses))) return {"status": "completed" if code == 0 else "failed", "returncode": code} return lambda *args, **kwargs: worker.run_supervised(*args, **kwargs, execute=execute) @@ -161,6 +161,80 @@ def test_unknown_usage_retains_attempt_and_does_not_disclose_other_arm(staged): assert summary["requests"]["A2"][0]["status"] == "unresolved" assert summary["requests"]["A2"][0]["usage"] is None assert staged.world.pr.ledger.charged == 162720000 + assert summary["diagnostics"]["A2"]["reason"] == "unknown_usage" + assert summary["diagnostics"]["A2"]["provider_attempts"] == 1 + assert summary["diagnostics"]["A3"]["provider_attempts"] is None + + +def test_closed_worker_failure_is_packaged_without_retry_or_arbitrary_text(staged): + seen = [] + transport = Mock(side_effect=provider.AgentError("http_429")) + job.execute(staged.output, api=SimpleNamespace(), + stores=(staged.world.daily, staged.world.pr), + launch=launcher([], seen, transport=transport)) + assert seen == ["A2"] and transport.call_count == 1 + assert staged.world.pr.ledger.charged == 162720000 + summary = job.package(staged.output) + diagnostic = summary["diagnostics"]["A2"] + assert diagnostic == {"worker_status": "failed", "reason": "provider_transport_error", + "provider_attempts": 1, "transport_error_code": "http_429", + "execution": {"status": "failed", "returncode": 1, "elapsed_seconds": None}} + assert summary["requests"]["A2"][0]["usage"] is None + assert summary["diagnostics"]["A3"]["worker_status"] == "unavailable" + public = b"".join(p.read_bytes() for p in (staged.output / "public").rglob("*") if p.is_file()) + assert b"OFFLINE-FAKE-PROVIDER-KEY" not in public + + +def test_failed_request_gate_consumes_slot_but_retains_zero_transport_invocations( + staged, monkeypatch): + monkeypatch.setattr(job.admission._Admission, "authorize", lambda *_: False) + seen, transport = [], Mock() + result = job.execute(staged.output, api=SimpleNamespace(), + stores=(staged.world.daily, staged.world.pr), + launch=launcher([], seen, transport=transport)) + assert seen == ["A2"] and transport.call_count == 0 + assert result["arms"]["A2"]["slots_consumed"] == 1 + assert result["arms"]["A2"]["request_completion_known"] is False + summary = job.package(staged.output) + assert summary["diagnostics"]["A2"]["provider_attempts"] == 0 + assert summary["diagnostics"]["A2"]["reason"] == "request_authorization_denied" + assert summary["diagnostics"]["A3"]["provider_attempts"] is None + assert staged.world.pr.ledger.charged == 162720000 + + +@pytest.mark.parametrize("attempts", [0, 1, 8, True, -1, 9, "1", None]) +def test_closed_outcome_recovers_fixed_files_even_without_execution_summary( + staged, attempts): + root = staged.output / "A2/run" + (root / "worker").mkdir(parents=True) + atomic_write_json(root / "worker/result.json", {"schema": worker.SCHEMA, + "status": "failed", "reason": "PRIVATE-CANARY arbitrary text", + "provider_attempts": attempts, "transport_error_code": "PRIVATE-CANARY", + "environment": {"secret": "PRIVATE-CANARY"}}) + atomic_write_json(root / "supervision.json", {"schema": "kv-worker-supervision-v1", + "execution": {"status": "failed", "returncode": -15, "elapsed_seconds": 1.5, + "metric": 99, "argv": ["PRIVATE-CANARY"], "error": "PRIVATE-CANARY"}}) + assert not (staged.output / "execution.json").exists() + summary = job.package(staged.output) + diagnostic = summary["diagnostics"]["A2"] + assert diagnostic["reason"] == diagnostic["transport_error_code"] == "unknown" + assert diagnostic["provider_attempts"] == ( + attempts if type(attempts) is int and 0 <= attempts <= 8 else None) + assert diagnostic["execution"] == { + "status": "failed", "returncode": -15, "elapsed_seconds": 1.5} + assert summary["diagnostics"]["A3"]["provider_attempts"] is None + assert "PRIVATE-CANARY" not in provider.canonical(summary) + assert summary["artifacts"]["A2"]["status"] == "missing_or_invalid" + + +@pytest.mark.parametrize("data", [b"not-json private", b"[]", b'{"schema":"wrong"}']) +def test_missing_or_malformed_worker_result_does_not_infer_zero(staged, data): + root = staged.output / "A2/run" + write(root, "worker/result.json", data) + diagnostic = job.package(staged.output)["diagnostics"]["A2"] + assert diagnostic == {"worker_status": "unavailable", "reason": None, + "provider_attempts": None, "transport_error_code": None, + "execution": {"status": "unavailable", "returncode": None, "elapsed_seconds": None}} def test_prepared_input_mutation_blocks_before_reservation(staged): diff --git a/tests/test_kv_live_worker.py b/tests/test_kv_live_worker.py index 663ad486b..802290b8b 100644 --- a/tests/test_kv_live_worker.py +++ b/tests/test_kv_live_worker.py @@ -1,13 +1,17 @@ """Synthetic-key/fake-HTTP controls only; never provider calls or live ledger access.""" import copy +import io +import json import time from unittest.mock import Mock +from urllib.error import HTTPError import pytest from inferrsi.model import anthropic_agent as provider from tests.test_anthropic_edit_agent import read_task +from tests.test_kv_pilot_admission import policy as sample_policy from tools import kv_live_worker as worker GATE = {"SKILL_EVAL_GATE_PORT": "1234", "SKILL_EVAL_GATE_TOKEN": "a" * 64, @@ -105,3 +109,58 @@ def test_ninth_authorized_send_cannot_make_a_ninth_http_attempt(tmp_path): with pytest.raises(provider.AgentError): value.send(payload, value.config["client"], 100) assert http.call_count == 8 + + +@pytest.mark.parametrize("failure,expected", [ + (401, "http_401"), (429, "http_429"), (500, "http_500"), (503, "http_503"), + ("bad_json", "invalid_response_json"), ("echo", "credential_echo"), + ("network", "transport_error"), +]) +def test_real_transport_closed_failure_code_survives_generic_failure( + tmp_path, monkeypatch, failure, expected): + value, _, _ = sender(tmp_path, transport=provider.send_message) + value.config["client"] = json.loads(sample_policy().provider_json)["client"] + opener = Mock() + monkeypatch.setattr(provider.urllib.request, "build_opener", Mock(return_value=opener)) + if type(failure) is int: + opener.open.side_effect = HTTPError("https://unused.invalid", failure, + "OFFLINE-FAKE-KEY private error", {}, io.BytesIO(b"private response body")) + elif failure == "network": + opener.open.side_effect = OSError("OFFLINE-FAKE-KEY private network detail") + else: + raw = b"not JSON: private response" if failure == "bad_json" else ( + provider.canonical({"text": "OFFLINE-FAKE-KEY"}).encode()) + opener.open.return_value = io.BytesIO(raw) + payload = {"model": "claude-opus-5-5"} + value.authorize(grant(payload), value.deadline) + with pytest.raises(provider.AgentError, match="^provider_transport_error$"): + value.send(payload, value.config["client"], 100) + assert value.transport_error_code == expected + assert value.attempts == opener.open.call_count == 1 + assert worker.scripted._read_json(tmp_path / "complete.json")["usage"] is None + with pytest.raises(provider.AgentError): + value.send(payload, value.config["client"], 100) + assert opener.open.call_count == 1 + + +def test_arbitrary_exception_and_nonclosed_codes_are_never_retained(tmp_path): + value, http, _ = sender(tmp_path, transport=Mock( + side_effect=provider.AgentError("OFFLINE-FAKE-KEY http_401 private payload"))) + payload = {"model": "claude-opus-5-5"} + value.authorize(grant(payload), value.deadline) + with pytest.raises(provider.AgentError, match="^provider_transport_error$"): + value.send(payload, value.config["client"], 100) + assert value.transport_error_code == "unknown" + assert value.attempts == http.call_count == 1 + for invalid in ("http_099", "http_600", "http_４０１", "http_401\n", {}, None, True): + assert worker.closed_reason(invalid) == "unknown" + + +def test_failed_presend_gate_has_known_zero_transport_invocations(tmp_path): + value, http, ipc = sender(tmp_path) + ipc.side_effect = OSError("OFFLINE-FAKE-KEY private gate detail") + with pytest.raises(OSError): + value.authorize(grant({"model": "claude-opus-5-5"}), value.deadline) + assert value.attempts == http.call_count == 0 + assert value.transport_error_code is None + assert not (tmp_path / "complete.json").exists() diff --git a/tools/kv_live_job.py b/tools/kv_live_job.py index a55db8dbc..5956f76d1 100644 --- a/tools/kv_live_job.py +++ b/tools/kv_live_job.py @@ -771,6 +771,39 @@ def _grade_evidence(bundle, output, result, plan=None): return manifest +def _worker_outcome(root): + """Recover only closed trusted outcome fields, even if controller writing stopped.""" + outcome = {"worker_status": "unavailable", "reason": None, "provider_attempts": None, + "transport_error_code": None, + "execution": {"status": "unavailable", "returncode": None, + "elapsed_seconds": None}} + try: + value = policy.read_json(root / "worker/result.json") + if type(value) is not dict or value.get("schema") != worker.SCHEMA: + raise ValueError("Wrong worker result") + if value.get("status") in {"completed", "failed", "inconclusive"}: + outcome["worker_status"] = value["status"] + if "reason" in value: + outcome["reason"] = worker.closed_reason(value["reason"]) + attempts = value.get("provider_attempts") + if type(attempts) is int and 0 <= attempts <= 8: + outcome["provider_attempts"] = attempts + if value.get("transport_error_code") is not None: + outcome["transport_error_code"] = worker.closed_reason(value["transport_error_code"]) + except (OSError, ValueError, KeyError, TypeError): + pass + try: + value = policy.read_json(root / "supervision.json") + if type(value) is not dict or value.get("schema") != "kv-worker-supervision-v1": + raise ValueError("Wrong supervision result") + projected = _execution_projection(value.get("execution")) + outcome["execution"].update({k: v for k, v in projected.items() + if k in outcome["execution"]}) + except (OSError, ValueError, KeyError, TypeError): + pass + return outcome + + def package(output): """Fixed bounded public projection. Never copy raw worker/provider logs or private paths.""" output = Path(output) @@ -791,6 +824,8 @@ def package(output): if (output / "execution.json").is_file() else {}) summary = {"schema": "kv-live-bundle-v1", "execution": execution, "artifacts": {}, "requests": {}, "evidence": {}, "preflight_status": preflight_status} + summary["diagnostics"] = {arm: _worker_outcome(output / arm / "run") + for arm in ("A2", "A3")} for arm in ("A2", "A3"): try: supervision = policy.read_json(output / arm / "run/supervision.json") @@ -819,7 +854,7 @@ def package(output): summary["evidence"][arm] = _public_evidence(root, public / arm, arm, execution) try: result = policy.read_json(root / "result.json") - if result.get("status") != "completed": + if type(result) is not dict or result.get("status") != "completed": raise ValueError("not submitted") receipt = policy.read_json(root / "final/receipt.json", 16384) data = kv.source_bytes(root / "final/candidate.py") diff --git a/tools/kv_live_worker.py b/tools/kv_live_worker.py index 4cc61d44a..3f200fbf2 100644 --- a/tools/kv_live_worker.py +++ b/tools/kv_live_worker.py @@ -27,10 +27,32 @@ from tools.kv_worker_supervisor import supervise # noqa: E402 SCHEMA = "kv-live-worker-v1" +REASONS = frozenset({ + "unknown", "sandbox_unavailable", "live_worker_error", "provider_transport_error", + "transport_error", "invalid_response_json", "response_bytes_limit", "credential_echo", + "invalid_credential_shape", "credential_in_payload", "request_authorization_denied", + "request_without_exact_grant", "request_bytes_limit", "timeout", "unknown_usage", + "incompatible_response", "input_token_limit", "output_token_limit", "request_limit", + "request_allocation_unavailable", "workspace_symlink", "invalid_workspace_file", + "tool_read_limit", "protected_workspace_changed", "unexposed_path", "invalid_candidate", + "candidate_bytes_limit", "invalid_visible_feedback", "feedback_bytes_limit", + "revision_retention_error", "deadline_exceeds_frozen_limit", "tool_call_limit", + "invalid_tool_call", "unsupported_tool", "task_not_read", "visible_callback_error", + "stale_or_missing_receipt", "edit_agent_error", "feedback_retention_limit", +}) FIELDS = {"schema", "arm", "model", "client", "edit_plan", "input_token_ceiling", "suite_sha256", "baseline_sha256", "actor_sha256", "dependencies", "prompt"} +def closed_reason(value): + """Fixed codes only; never exception text, provider bodies or credentials.""" + if type(value) is str and (value in REASONS or ( + len(value) == 8 and value.startswith("http_") and value[5:].isascii() + and value[5:].isdigit() and 100 <= int(value[5:]) <= 599)): + return value + return "unknown" + + def dependencies(): return {**scripted.dependencies(), "tools/kv_live_worker.py": edit.digest(Path(__file__).read_bytes())} @@ -89,6 +111,7 @@ def __init__(self, config, bridge, deadline, *, key, gate, self.key, self.gate = key, gate_environment(gate) self.transport, self.authorize_ipc = transport, authorize_ipc self.grant, self.ready, self.sent, self.attempts = None, False, False, 0 + self.transport_error_code = None def authorize(self, grant, deadline): if (deadline != self.deadline or time.monotonic() >= deadline @@ -115,7 +138,11 @@ def send(self, payload, spec, remaining): # prevents send_message's implicit environment gate from double-counting. value = self.transport(payload, spec, self.key, min(remaining, left), environment={}, require_authorization=False) - except Exception: + except Exception as error: + # Keep a separate closed diagnostic before the edit agent generalizes + # transport failures. Counting this invocation does not prove HTTP delivery. + self.transport_error_code = closed_reason(error.args[0] if ( + isinstance(error, provider.AgentError) and len(error.args) == 1) else None) atomic_write_json(self.bridge / "complete.json", {"grant": self.grant, "usage": None}) raise provider.AgentError("provider_transport_error") from None atomic_write_json(self.bridge / "complete.json", {"grant": self.grant, @@ -176,10 +203,11 @@ def retain_feedback(event): except kv.SandboxUnavailable: result.update(status="inconclusive", reason="sandbox_unavailable") except provider.AgentError as error: - result["reason"] = str(error) + result["reason"] = closed_reason(error.args[0] if len(error.args) == 1 else None) except Exception: result["reason"] = "live_worker_error" result["provider_attempts"] = sender.attempts if sender else 0 + result["transport_error_code"] = sender.transport_error_code if sender else None atomic_write_json(output / "result.json", result) return 0 if result["status"] == "completed" else 1