Retain closed KV failure diagnostics without retry

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

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):

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()

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")

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