feat(harness): agent reading, failure record and tool routing observations for readiness

CHANGELOG.md

@@ -25,6 +25,11 @@ Post-rc.1 increments on the release branch. Each one adds gates on top of the
   and the person's own session records what the agent actually heard.
 - Runtime validation checks scenarios on one database lane per CPU by default, capped by the
   sandbox's cgroup CPU quota, instead of a fixed 16; `ALK_VALIDATION_DB_LANES` still overrides it.
+- Chat readiness tool routing: the tool proxy logs each call's full path, so two service tools
+  whose paths share a last segment are told apart and an older record naming only that segment
+  credits neither; a tool request's handler is bounded by the request deadline and a handler that
+  outlasts it leaves the tool unobserved; an earlier answer is reused only for the same candidate
+  and tool definition, within the one request and time budget for the check.
 - Docs corpus: the full `docs/` tree (quickstarts, per-track guides, framework
   pages, reference material, and the `docs/llms.txt` machine index) with
   machine-checkable page metadata, enforced by the new `docs_executability`

src/fi/alk/harness/authoring_runtime_validation.py

@@ -20,6 +20,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from . import failure_record
+
 if TYPE_CHECKING:
     from .source_model import SourceModel
     from .world_ir import WorldIR
@@ -330,7 +332,39 @@ def _make_local_seed_files_readable(bundle: Path) -> None:
         path.chmod(path.stat().st_mode | (0o055 if path.is_dir() else 0o044))
 
 
-def _write_generic_certificate(job, authoring: Path, repairs) -> None:
+def _has_judged_sub_goals(authoring: Path) -> bool:
+    from .catalogue import load_catalogue
+
+    for root in (authoring, authoring / "environment-bundle"):
+        try:
+            catalogue = load_catalogue(root)
+        except (OSError, ValueError):
+            continue
+        if any(goal.judged.strip() for goal in catalogue.sub_goals):
+            return True
+    return False
+
+
+def _with_judge_check(checks, authoring: Path):
+    """``checks`` with whether the judge answered, only when a judged sub-goal needs it."""
+    from .certification import CheckStatus
+    from .judge import judge_reachable
+
+    if not _has_judged_sub_goals(authoring):
+        return checks
+    return checks.model_copy(update={"judge_reachable": CheckStatus(judge_reachable())})
+
+
+def _harness_version() -> str | None:
+    from fi.alk import __version__
+
+    return None if __version__.startswith("0.0.0+") else __version__
+
+
+def _write_generic_certificate(
+    job, authoring: Path, repairs, source: Path | None = None
+) -> None:
+    from .generated_runtime import installed_frameworks
     from .certification import (
         CertificationAuthoring,
         CertificationCompiler,
@@ -343,12 +377,12 @@ def _write_generic_certificate(job, authoring: Path, repairs) -> None:
 
     store = GenericHarnessArtifactStore(authoring / "generic-harness")
     evidence = store.read_runtime_evidence()
-    source = getattr(job, "source", None)
+    origin = getattr(job, "source", None)
     certificate = HarnessCertification.create(
         status=CertificationStatus.CERTIFIED,
         source=CertificationSource(
-            repository=getattr(source, "repository", None),
-            commit=getattr(source, "commit_sha", None),
+            repository=getattr(origin, "repository", None),
+            commit=getattr(origin, "commit_sha", None),
             digest=evidence.source_digest,
             schema_hash=evidence.source_schema_hash,
         ),
@@ -364,8 +398,10 @@ def _write_generic_certificate(job, authoring: Path, repairs) -> None:
         runtime=CertificationRuntime(
             snapshot=_runtime_snapshot(job),
             validation_attempts=len(repairs.decisions),
+            harness_version=_harness_version(),
+            frameworks=installed_frameworks(source),
         ),
-        checks=evidence.checks,
+        checks=_with_judge_check(evidence.checks, authoring),
         repairs=repairs,
         limitations=evidence.limitations,
     )
@@ -1178,7 +1214,13 @@ async def check_setup_once(scenario, invariants, against, keep_processes):
                     _tools,
                     _chat_target_timeout_seconds,
                 )
-                from .chat_readiness import probe_chat_ingress, probe_chat_workflow
+                from .chat_readiness import (
+                    probe_chat_ingress,
+                    probe_chat_workflow,
+                    record_tool_routing,
+                    request_uncovered_tools,
+                    world_call_log,
+                )
 
                 chat_ingress = CheckStatus.NOT_APPLICABLE
                 interface = contract.runtime.interface if contract.runtime else None
@@ -1217,7 +1259,20 @@ async def check_setup_once(scenario, invariants, against, keep_processes):
                             secret_values=target_secret_values,
                             redaction_values=redaction_values,
                         )
+                        await request_uncovered_tools(
+                            interface=interface,
+                            endpoint=endpoint.address,
+                            runtime=runtime,
+                            contract=contract,
+                            authoring=authoring,
+                            bundle=bundle,
+                            source=source,
+                            secret_values=target_secret_values,
+                            candidate_hash=_generic_candidate_hash(source, authoring),
+                            local_auth_values=getattr(runtime, "_local_auth_values", None),
+                        )
                     else:
+                        since = time.time()
                         await probe_chat_ingress(
                             interface=interface,
                             endpoint=endpoint.address,
@@ -1233,6 +1288,34 @@ async def check_setup_once(scenario, invariants, against, keep_processes):
                             scenario_id=representative.scenario_id,
                             timeout=_chat_target_timeout_seconds(),
                         )
+                        from .call_runner import _collect_file_tool_calls
+
+                        # A test message can call tools too; then ask for the rest.
+                        try:
+                            called = _collect_file_tool_calls(runtime)
+                        except Exception:  # noqa: BLE001 - no trace, nothing observed
+                            called = ()
+                        record_tool_routing(
+                            authoring,
+                            representative.scenario_key,
+                            contract,
+                            runtime,
+                            received=world_call_log(runtime, since),
+                            reported=called,
+                            bundle=bundle,
+                        )
+                        await request_uncovered_tools(
+                            interface=interface,
+                            endpoint=endpoint.address,
+                            runtime=runtime,
+                            contract=contract,
+                            authoring=authoring,
+                            bundle=bundle,
+                            source=source,
+                            secret_values=target_secret_values,
+                            candidate_hash=_generic_candidate_hash(source, authoring),
+                            local_auth_values=getattr(runtime, "_local_auth_values", None),
+                        )
                     chat_ingress = CheckStatus.PASSED
             phase = "environment"
             await provider.reset(runtime, work_directory=work)
@@ -1626,6 +1709,8 @@ def write_history():
         )
         candidate_hash = _generic_candidate_hash(source, authoring)
         for _attempt in range(decision_bound):
+            # A record left by an earlier attempt must never speak for this one.
+            failure_record.clear()
             try:
                 count = await validate(job, source, authoring)
             except RuntimeValidationError as exc:
@@ -1642,6 +1727,7 @@ def write_history():
                     f"runtime validation: {exc.phase}: {exc}",
                     flush=True,
                 )
+                failure_record.record(exc.phase, exc)
                 phase = (
                     RepairPhase.SCENARIOS
                     if exc.phase == "scenarios"
@@ -1875,7 +1961,7 @@ def write_history():
                 )
                 if evidence_path.is_file():
                     _write_generic_certificate(
-                        job, authoring, repair_controller.history
+                        job, authoring, repair_controller.history, source
                     )
                 else:
                     # Injected validators in unit/integration consumers may not implement the
@@ -1902,10 +1988,12 @@ def write_history():
     repairs = {"environment": 0, "scenarios": 0}
     for attempt in range(5):
         print(f"runtime validation: attempt {attempt + 1}/5", flush=True)
+        failure_record.clear()
         try:
             count = await validate(job, source, authoring)
         except RuntimeValidationError as exc:
             print(f"runtime validation: {exc.phase}: {exc}", flush=True)
+            failure_record.record(exc.phase, exc)
             if exc.phase not in repairs or repairs[exc.phase] >= 2:
                 raise
             repairs[exc.phase] += 1

src/fi/alk/harness/bundle_author_v2.py

@@ -3654,6 +3654,7 @@ def author_bundle_v2(
             "CREATE TABLE IF NOT EXISTS _alk_tool_trace ("
             "id bigserial PRIMARY KEY, name text NOT NULL, arguments jsonb NOT NULL, "
             "result jsonb, ok boolean NOT NULL, error text, at double precision NOT NULL);\n"
+            "ALTER TABLE _alk_tool_trace ADD COLUMN IF NOT EXISTS endpoint text;\n"
         )
         generic_pipeline = job.metadata.get("generic_harness_v1") is True
         if generic_pipeline:

src/fi/alk/harness/call_runner.py

@@ -745,7 +745,8 @@ def _collect_tool_trace_calls(runtime: EnvironmentRuntime) -> tuple[Call, ...]:
             options="-c default_transaction_read_only=on",
         ) as connection:
             cursor = connection.execute(
-                f'SELECT name, arguments, result, ok, error, at FROM "{_TOOL_TRACE_TABLE}" '  # noqa: S608
+                # Every column: an older table has no endpoint column.
+                f'SELECT * FROM "{_TOOL_TRACE_TABLE}" '  # noqa: S608
                 "ORDER BY at ASC"
             )
             rows = cursor.fetchall()
@@ -781,6 +782,7 @@ def _collect_tool_trace_calls(runtime: EnvironmentRuntime) -> tuple[Call, ...]:
                 error=error,
                 refused=not ok,
                 at=at,
+                endpoint=str(record.get("endpoint") or ""),
             )
         )
     return tuple(calls)

src/fi/alk/harness/certification.py

@@ -65,6 +65,21 @@ class CertificationRuntime(BaseModel):
 
     snapshot: str | None = None
     validation_attempts: int = Field(ge=1)
+    # The harness that made the certificate, and the agent framework distributions installed in the
+    # agent's environment, by version. Recorded only when known.
+    harness_version: str | None = None
+    frameworks: dict[str, str] = Field(default_factory=dict)
+
+    @model_serializer(mode="wrap")
+    def _without_unrecorded_versions(self, handler: Any) -> Any:
+        # A certificate made before these were recorded keeps its fingerprint and format.
+        data = handler(self)
+        if isinstance(data, dict):
+            if self.harness_version is None:
+                data.pop("harness_version", None)
+            if not self.frameworks:
+                data.pop("frameworks", None)
+        return data
 
 
 class CertificationChecks(BaseModel):
@@ -80,6 +95,9 @@ class CertificationChecks(BaseModel):
     action_behavior: str = "0/0"
     reset_equivalence: CheckStatus = CheckStatus.NOT_RUN
     world_isolation: CheckStatus = CheckStatus.NOT_RUN
+    # Whether the judge answered one short request during the check; recorded only when the
+    # environment has a judged sub-goal.
+    judge_reachable: CheckStatus | None = None
 
     @property
     def records_chat_ingress(self) -> bool:
@@ -92,6 +110,8 @@ def _without_unrecorded_chat_ingress(self, handler: Any) -> Any:
         data = handler(self)
         if not self.records_chat_ingress and isinstance(data, dict):
             data.pop("chat_ingress", None)
+        if self.judge_reachable is None and isinstance(data, dict):
+            data.pop("judge_reachable", None)
         return data
 
 

src/fi/alk/harness/chat_call_runner.py

@@ -330,6 +330,8 @@ def __init__(
         self._messages: list[dict[str, Any]] = []
         self._tool_call_ids: set[str] = set()
         self._turn = 0
+        # Calls the harness ran at the world itself, not ones the agent says it ran.
+        self.received: list[Call] = []
 
     async def open(self) -> None:
         return
@@ -411,6 +413,7 @@ async def say(self, utterance: str) -> str:
                     result = self.world.handle_tool_call(
                         {"id": call_id, "name": name, "arguments": arguments}
                     )
+                    self.received.append(self.world.calls[-1])
                     result_content = (
                         result.content if result is not None else f"no such tool {name}"
                     )
@@ -480,6 +483,8 @@ class HostedChatCallRunner:
     def __init__(self, adapter: ArtifactUploader, context: CallRunnerContext) -> None:
         self._adapter = adapter
         self._context = context
+        # The last conversation's calls the harness ran at the world (``_HostedChatTarget``).
+        self.received: list[Call] = []
         contract_path = context.bundle_dir / "contract.json"
         if not contract_path.is_file():
             self._contract: AgentContract | None = None
@@ -550,15 +555,17 @@ async def run(
         transcript: Transcript | None = None
         failure: CallAborted | None = None
         _clear_file_tool_calls(runtime)
+        target = _HostedChatTarget(
+            wrapper=wrapper,
+            contract=self._contract,
+            world=target_world,
+            scenario_key=scenario.scenario_key,
+            scenario_id=scenario.scenario_id,
+        )
+        self.received = target.received
         try:
             transcript = await _drive_conversation(
-                _HostedChatTarget(
-                    wrapper=wrapper,
-                    contract=self._contract,
-                    world=target_world,
-                    scenario_key=scenario.scenario_key,
-                    scenario_id=scenario.scenario_id,
-                ),
+                target,
                 conversation_scenario,
                 self._contract,
                 self._context.bundle_dir,