Seal per-scenario evidence so judged sub-goals can be decided off-sandbox

Dockerfile.hosted

@@ -72,7 +72,7 @@ WORKDIR /opt/alk
 # Daytona's direct-image builder can reuse an image when only build-context
 # files change.  Bump this source revision whenever guest code changes so a
 # hosted test cannot silently execute an older installed ALK.
-ARG ALK_HOSTED_SOURCE_REVISION=20260829-visible-diagnostics-r12
+ARG ALK_HOSTED_SOURCE_REVISION=20260901-evidence-artifact-r1
 LABEL io.futureagi.alk-source-revision="${ALK_HOSTED_SOURCE_REVISION}"
 COPY pyproject.toml README.md ./
 COPY src ./src

src/fi/alk/harness/hosted_entrypoint.py

@@ -657,6 +657,7 @@ def _cap_failure_message(message: str) -> str:
             ob.ArtifactKind.TRACE,
             ob.ArtifactKind.TOOL_TRACE,
             ob.ArtifactKind.TRANSCRIPT,
+            ob.ArtifactKind.EVIDENCE,
             ob.ArtifactKind.OTHER,
         }
     ),
@@ -1060,6 +1061,17 @@ async def receipt(self, receipt: ResultReceipt) -> None:
                 "code": receipt.failure.code,
                 "message": _cap_failure_message(redacted_failure_message),
             }
+        if receipt.evidence is not None:
+            # Before the receipt, like every other artifact it references: the platform judges
+            # judged sub-goals from this, so a receipt arriving first would be judged blind.
+            await self.upload_artifact(
+                json.dumps(
+                    receipt.evidence, ensure_ascii=False, sort_keys=True, default=str
+                ).encode("utf-8"),
+                kind=ob.ArtifactKind.EVIDENCE,
+                scenario_key=receipt.scenario_key,
+                deadline=self.deadline(),
+            )
         wire = ob.build_result_receipt(
             job_id=self._capabilities.job_id,
             attempt_id=self._capabilities.attempt_id,

src/fi/alk/harness/hosted_scheduler.py

@@ -32,6 +32,7 @@
 
 import asyncio
 import inspect
+import json
 import logging
 import random
 import re
@@ -309,6 +310,9 @@ class ResultReceipt:
     evaluations: tuple[Evaluation, ...]
     call: CallSummary | None
     failure: ReceiptFailure | None
+    # Uploaded as an `evidence` artifact by the emitter, never sent inline: the wire receipt is
+    # digest-signed and a world snapshot does not belong inside it.
+    evidence: dict[str, Any] | None = None
 
 
 # --- outbound (this module's own minimal sink; see the module docstring's decoupling note) ----
@@ -453,6 +457,140 @@ def _truncate(text: str, limit: int = _MESSAGE_LIMIT) -> str:
     return text[: limit - len("…[truncated]")] + "…[truncated]"
 
 
+EVIDENCE_SCHEMA_VERSION = "futureagi.harness-evidence.v1"
+# The judge runs on the platform, so the world it reasons about is whatever the guest serialized
+# here: this is the only moment the final state exists. Bounded because a seeded world is
+# unbounded, and every drop is counted rather than silent -- a judge told "40 of 340 rows" can
+# reason about the gap, one shown 40 and told nothing cannot.
+_EVIDENCE_ROWS_PER_TABLE = 40
+_EVIDENCE_MAX_BYTES = 128_000
+_EVIDENCE_MAX_CALLS = 200
+_EVIDENCE_CELL_LIMIT = 500
+
+
+def _jsonable(value: object, limit: int = _EVIDENCE_CELL_LIMIT) -> object:
+    if value is None or isinstance(value, (bool, int, float)):
+        return value
+    if isinstance(value, str):
+        return _truncate(value, limit)
+    if isinstance(value, dict):
+        return {str(key): _jsonable(item, limit) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_jsonable(item, limit) for item in value]
+    return _truncate(str(value), limit)
+
+
+def _relates(stem: str, text: str) -> bool:
+    """Whether a table stem and a piece of call text are about the same thing.
+
+    Prefix matching in both directions, per underscore-separated word: `book_ride` is evidence
+    about `bookings` (`book` prefixes `booking`), and `rider_id` about `riders`. Containment
+    alone misses the first of those, which is the common case.
+    """
+    for word in re.split(r"[^a-z0-9]+", text.lower()):
+        if len(word) < 3:
+            continue
+        if stem.startswith(word) or word.startswith(stem):
+            return True
+    return False
+
+
+def _table_relevance(table: str, calls: Sequence[Call]) -> int:
+    """How strongly the calls point at this table, so the byte budget is spent where the run acted."""
+    stem = table.lower().rstrip("s")
+    if len(stem) < 3:
+        return 0
+    score = 0
+    for call in calls:
+        if _relates(stem, call.name):
+            score += 2
+        for key, value in (call.arguments or {}).items():
+            if _relates(stem, str(key)) or _relates(stem, str(value)):
+                score += 1
+    return score
+
+
+def _capture_evidence(
+    scenario_key: str,
+    world: ReadOnlyWorld,
+    calls: Sequence[Call],
+    sub_goals: Sequence[SubGoal] = (),
+) -> dict[str, Any]:
+    """The run's final state, for a judge that will never see the world itself.
+
+    Carries each judged sub-goal's claim too: the platform knows a sub-goal was judged but not
+    what it was meant to decide, which lives only in the bundle catalogue on this side.
+    """
+    claims = [
+        {
+            "name": goal.name,
+            "what": str(getattr(goal, "what", "") or ""),
+            "judged": goal.judged,
+        }
+        for goal in sub_goals
+        if goal.judged
+    ]
+    recorded = [
+        {
+            "name": call.name,
+            "arguments": _jsonable(call.arguments or {}),
+            "result": _jsonable(call.result),
+            "ok": bool(call.ok),
+            "error": _truncate(str(call.error or ""), _EVIDENCE_CELL_LIMIT),
+            "refused": bool(call.refused),
+            "at": call.at,
+        }
+        for call in list(calls)[:_EVIDENCE_MAX_CALLS]
+    ]
+    try:
+        state = world.state()
+    except Exception as exc:  # noqa: BLE001 - evidence is best effort, never a failed scenario
+        return {
+            "schema_version": EVIDENCE_SCHEMA_VERSION,
+            "scenario_key": scenario_key,
+            "calls": recorded,
+            "calls_total": len(calls),
+            "judged_sub_goals": claims,
+            "world": {"unavailable": _sanitize_cause(f"{type(exc).__name__}: {exc}")},
+        }
+
+    ordered = sorted(
+        state.items(),
+        key=lambda item: (
+            -_table_relevance(item[0], calls),
+            len(item[1] or []),
+            item[0],
+        ),
+    )
+    tables: dict[str, Any] = {}
+    omitted: list[str] = []
+    budget = _EVIDENCE_MAX_BYTES
+    for name, rows in ordered:
+        rows = list(rows or [])
+        if budget <= 0:
+            omitted.append(name)
+            continue
+        shown = [_jsonable(row) for row in rows[:_EVIDENCE_ROWS_PER_TABLE]]
+        cost = len(json.dumps(shown, default=str))
+        if cost > budget and tables:
+            omitted.append(name)
+            continue
+        budget -= cost
+        tables[name] = {
+            "rows": shown,
+            "rows_shown": len(shown),
+            "rows_total": len(rows),
+        }
+    return {
+        "schema_version": EVIDENCE_SCHEMA_VERSION,
+        "scenario_key": scenario_key,
+        "calls": recorded,
+        "calls_total": len(calls),
+        "judged_sub_goals": claims,
+        "world": {"tables": tables, "omitted_tables": omitted},
+    }
+
+
 def _sanitize_cause(message: str) -> str:
     # M7: `cause` is capped at 200 chars and must never carry endpoint credentials — postgres
     # error strings routinely embed the DSN (`postgresql://user:pw@host/db`).
@@ -2020,6 +2158,9 @@ async def _execute(
             evaluations=(),
             call=self._call_summary(call_outcome),
             failure=None,
+            evidence=_capture_evidence(
+                scenario.scenario_key, check_handle, calls, scenario.sub_goals
+            ),
         )
 
     @staticmethod

src/fi/alk/harness/outbound.py

@@ -2760,6 +2760,7 @@ class ArtifactKind(str, Enum):
     RECORDING_ASSISTANT = "recording_assistant"
     TRANSCRIPT = "transcript"
     TOOL_TRACE = "tool_trace"
+    EVIDENCE = "evidence"
     RESULT = "result"
     BUILD = "build"
     TRACE = "trace"
@@ -2772,6 +2773,7 @@ class ArtifactKind(str, Enum):
         ArtifactKind.BUILD,
         ArtifactKind.TRANSCRIPT,
         ArtifactKind.TOOL_TRACE,
+        ArtifactKind.EVIDENCE,
         ArtifactKind.RESULT,
     }
 )
@@ -2939,6 +2941,7 @@ class ManifestPushResult:
     ArtifactKind.RECORDING_CUSTOMER: "video/mp4",
     ArtifactKind.RECORDING_ASSISTANT: "video/mp4",
     ArtifactKind.TRANSCRIPT: "application/json",
+    ArtifactKind.EVIDENCE: "application/json",
     ArtifactKind.RESULT: "application/json",
 }
 

src/fi/alk/harness/scenario_source.py

@@ -42,6 +42,7 @@
 
 _CHECKS_DIRNAME = "checks"
 _SCENARIO_JSON = "scenario.json"
+_CATALOGUE_JSON = "sub_goals.json"
 _SETUP_PY = "setup.py"
 _READY_PY = "ready.py"
 
@@ -167,6 +168,7 @@ class _CompiledSubGoal:
     name: str
     judged: str
     check: Callable[[Any, Any], object]
+    what: str = ""
 
 
 @dataclass(frozen=True)
@@ -213,6 +215,31 @@ def _validate_subgoal_name(name: str, *, folder_name: str) -> None:
         )
 
 
+def _load_catalogue_claims(bundle_dir: Path) -> dict[str, dict[str, str]]:
+    """`sub_goals.json`'s `what`/`judged` text, which `folder.py` never writes into a scenario
+    folder. Without it a judged sub-goal reaches the platform as a name and nothing to decide.
+    """
+    path = bundle_dir / _CATALOGUE_JSON
+    if not path.is_file():
+        return {}
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
+        return {}
+    entries = raw.get("sub_goals") if isinstance(raw, dict) else None
+    if not isinstance(entries, list):
+        return {}
+    claims: dict[str, dict[str, str]] = {}
+    for entry in entries:
+        if not isinstance(entry, dict) or not isinstance(entry.get("name"), str):
+            continue
+        claims[entry["name"]] = {
+            "what": str(entry.get("what") or ""),
+            "judged": str(entry.get("judged") or ""),
+        }
+    return claims
+
+
 def _load_one(folder: Path) -> _CompiledScenario:
     """One scenario folder -> a `Scenario`-protocol object. Mirrors `folder.py`'s documented
     layout (`scenario.json` + `setup.py` + `ready.py` + `checks/<goal>.py`) but reads
@@ -285,6 +312,30 @@ def _load_one(folder: Path) -> _CompiledScenario:
     )
 
 
+def _with_claims(
+    scenario: _CompiledScenario, claims: dict[str, dict[str, str]]
+) -> _CompiledScenario:
+    """Restore each judged sub-goal's real claim from the catalogue.
+
+    `_load_one` can only tell that a sub-goal is judged, never what it was meant to decide:
+    `folder.py` writes no file for one. Without this the platform judge gets a name and a
+    placeholder, which is not something a verdict can be reached from.
+    """
+    if not claims:
+        return scenario
+    restored = tuple(
+        replace(
+            goal,
+            judged=claims[goal.name].get("judged") or goal.judged,
+            what=claims[goal.name].get("what", ""),
+        )
+        if goal.judged and goal.name in claims
+        else goal
+        for goal in scenario.sub_goals
+    )
+    return replace(scenario, sub_goals=restored)
+
+
 def load_scenarios(bundle_dir: Path) -> list[_CompiledScenario]:
     """Every scenario document under `<bundle_dir>/scenarios/`, compiled and wrapped, in the same
     sorted-by-folder-name order `folder.py`'s `read_all` uses. Raises `ScenarioDocumentInvalid` on
@@ -302,11 +353,12 @@ def load_scenarios(bundle_dir: Path) -> list[_CompiledScenario]:
         # document (R1-1) -- this is inside `run_job`'s `try`/`except ScenarioDocumentInvalid`
         # (unlike `bundle_has_scenarios`'s own guard above), so raising here is the safe direction.
         raise ScenarioDocumentInvalid(f"{root}: cannot list scenario folders: {exc}") from exc
+    claims = _load_catalogue_claims(bundle_dir)
     scenarios: list[_CompiledScenario] = []
     for folder in entries:
         if not folder.is_dir():
             continue
-        scenarios.append(_load_one(folder))
+        scenarios.append(_with_claims(_load_one(folder), claims))
     if not scenarios:
         raise ScenarioDocumentInvalid(f"{root} contains no scenario folders")
     return scenarios

tests/harness/test_hosted_entrypoint.py

@@ -3676,3 +3676,58 @@ async def scenario() -> None:
             print(f"ok   {test_fn.__name__} ({time.monotonic() - started:.2f}s)")
     print(f"\n{len(TESTS) - failures}/{len(TESTS)} passed")
     raise SystemExit(1 if failures else 0)
+
+
+def test_receipt_seals_evidence_as_an_artifact_before_pushing_the_receipt() -> None:
+    # The wire receipt is digest-signed and a world snapshot does not belong inside it, so
+    # evidence rides as an `evidence` artifact keyed by scenario. It must be uploaded BEFORE the
+    # receipt, like every other artifact a receipt refers to: a receipt landing first would be
+    # judged against evidence the platform does not have yet.
+    async def scenario() -> None:
+        transport = FakeTransport()
+        adapter = _build_adapter(transport)
+        await adapter.receipt(
+            ResultReceipt(
+                scenario_key="s1",
+                scenario_id="platform-s1",
+                scenario_attempt=1,
+                world_index=0,
+                status="passed",
+                sub_goals=(),
+                evaluations=(),
+                call=None,
+                failure=None,
+                evidence={"schema_version": "futureagi.harness-evidence.v1", "calls": []},
+            )
+        )
+        sealed = [json.loads(body) for body in transport.artifacts.values()]
+        assert any(
+            item.get("schema_version") == "futureagi.harness-evidence.v1"
+            for item in sealed
+        )
+        # And the receipt itself stays exactly the shape the platform already validates.
+        assert "evidence" not in transport.receipts[("job-1", "s1")]
+
+    asyncio.run(scenario())
+
+
+def test_receipt_without_evidence_uploads_nothing_extra() -> None:
+    async def scenario() -> None:
+        transport = FakeTransport()
+        adapter = _build_adapter(transport)
+        await adapter.receipt(
+            ResultReceipt(
+                scenario_key="s1",
+                scenario_id="platform-s1",
+                scenario_attempt=1,
+                world_index=0,
+                status="passed",
+                sub_goals=(),
+                evaluations=(),
+                call=None,
+                failure=None,
+            )
+        )
+        assert transport.artifacts == {}
+
+    asyncio.run(scenario())