From 1ef9ab60c09d9b340c0df99997904c566ab60fea Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Mon, 31 Aug 2026 23:33:40 +0530 Subject: [PATCH 1/3] feat(harness): seal per-scenario evidence so judged sub-goals can be decided off-sandbox --- Dockerfile.hosted | 2 +- src/fi/alk/harness/hosted_entrypoint.py | 12 ++ src/fi/alk/harness/hosted_scheduler.py | 141 ++++++++++++++++++++++++ src/fi/alk/harness/outbound.py | 3 + src/fi/alk/harness/scenario_source.py | 54 ++++++++- 5 files changed, 210 insertions(+), 2 deletions(-) diff --git a/Dockerfile.hosted b/Dockerfile.hosted index fbb836fd..a803ecb4 100644 --- a/Dockerfile.hosted +++ b/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 diff --git a/src/fi/alk/harness/hosted_entrypoint.py b/src/fi/alk/harness/hosted_entrypoint.py index b54d9852..6f614f76 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/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, diff --git a/src/fi/alk/harness/hosted_scheduler.py b/src/fi/alk/harness/hosted_scheduler.py index 51c1fd07..39f21e0d 100644 --- a/src/fi/alk/harness/hosted_scheduler.py +++ b/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 diff --git a/src/fi/alk/harness/outbound.py b/src/fi/alk/harness/outbound.py index 235c0b0e..a1853011 100644 --- a/src/fi/alk/harness/outbound.py +++ b/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", } diff --git a/src/fi/alk/harness/scenario_source.py b/src/fi/alk/harness/scenario_source.py index 80455f40..09f209a9 100644 --- a/src/fi/alk/harness/scenario_source.py +++ b/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/.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 `/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 From 69bb6c54163bae30be55ea6e716ffff960d7a02e Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Mon, 31 Aug 2026 23:39:32 +0530 Subject: [PATCH 2/3] test(harness): cover evidence capture bounds, claims and world-read failure --- tests/harness/test_hosted_scheduler.py | 96 ++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/tests/harness/test_hosted_scheduler.py b/tests/harness/test_hosted_scheduler.py index 86463beb..278ff601 100644 --- a/tests/harness/test_hosted_scheduler.py +++ b/tests/harness/test_hosted_scheduler.py @@ -2541,3 +2541,99 @@ def setup(world: Any) -> None: await pool.close() asyncio.run(scenario()) + + +# --- evidence capture (judged sub-goals are decided off-sandbox, from this) -------------------- + + +class _EvidenceWorld: + def __init__(self, state): + self._state = state + + def state(self, table=None): + return self._state + + +class _EvidenceGoal: + def __init__(self, name, judged="", what=""): + self.name = name + self.judged = judged + self.what = what + + def check(self, world, calls): + return None + + +def _evidence_call(name="book_ride", arguments=None): + from fi.alk.harness.world.runtime import Call + + return Call( + name=name, + arguments=arguments if arguments is not None else {"when": "10pm"}, + result={"ok": True}, + ok=True, + at=1.0, + ) + + +def test_evidence_carries_calls_with_arguments_and_judged_claims(): + from fi.alk.harness.hosted_scheduler import _capture_evidence + + evidence = _capture_evidence( + "s1", + _EvidenceWorld({"bookings": [{"id": 1, "when": "10pm"}]}), + [_evidence_call()], + [_EvidenceGoal("booked"), _EvidenceGoal("surge_told", "a judge must decide", "surge")], + ) + + assert evidence["calls"][0]["arguments"] == {"when": "10pm"} + # Only the judged one carries a claim: the platform knows a sub-goal was judged but not + # what it was meant to decide. + assert [claim["name"] for claim in evidence["judged_sub_goals"]] == ["surge_told"] + assert evidence["judged_sub_goals"][0]["judged"] == "a judge must decide" + + +def test_evidence_counts_the_rows_it_dropped(): + from fi.alk.harness.hosted_scheduler import _EVIDENCE_ROWS_PER_TABLE, _capture_evidence + + rows = [{"id": index} for index in range(_EVIDENCE_ROWS_PER_TABLE + 25)] + evidence = _capture_evidence( + "s1", _EvidenceWorld({"bookings": rows}), [_evidence_call()], [] + ) + + table = evidence["world"]["tables"]["bookings"] + assert table["rows_shown"] == _EVIDENCE_ROWS_PER_TABLE + assert table["rows_total"] == len(rows) + + +def test_evidence_spends_its_budget_on_the_tables_the_run_touched(): + from fi.alk.harness.hosted_scheduler import _capture_evidence + + evidence = _capture_evidence( + "s1", + _EvidenceWorld( + { + "audit_log": [{"id": index, "blob": "x" * 900} for index in range(500)], + "bookings": [{"id": 1}], + } + ), + [_evidence_call()], + [], + ) + + # `book_ride` is evidence about `bookings`, so it must not be crowded out by a large + # table nothing in the run referred to. + assert list(evidence["world"]["tables"])[0] == "bookings" + + +def test_evidence_survives_a_world_that_cannot_be_read(): + from fi.alk.harness.hosted_scheduler import _capture_evidence + + class _Gone: + def state(self, table=None): + raise RuntimeError("gone: postgresql://user:pw@host/db") + + evidence = _capture_evidence("s1", _Gone(), [_evidence_call()], []) + + assert evidence["calls"] + assert "pw" not in evidence["world"]["unavailable"] From 6fe947f51bca3a948dcbfa90e7a8fa224a4c6aea Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Mon, 31 Aug 2026 23:42:37 +0530 Subject: [PATCH 3/3] test(harness): assert evidence is sealed as an artifact and kept off the wire receipt --- tests/harness/test_hosted_entrypoint.py | 55 +++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/harness/test_hosted_entrypoint.py b/tests/harness/test_hosted_entrypoint.py index fe2690c1..7ea3ccda 100644 --- a/tests/harness/test_hosted_entrypoint.py +++ b/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())