diff --git a/CHANGELOG.md b/CHANGELOG.md index a846a177e..4498503b0 100644 --- a/CHANGELOG.md +++ b/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` diff --git a/src/fi/alk/harness/authoring_runtime_validation.py b/src/fi/alk/harness/authoring_runtime_validation.py index 8d608d511..2980c1728 100644 --- a/src/fi/alk/harness/authoring_runtime_validation.py +++ b/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 diff --git a/src/fi/alk/harness/bundle_author_v2.py b/src/fi/alk/harness/bundle_author_v2.py index 0ee5f0ada..9f74da337 100644 --- a/src/fi/alk/harness/bundle_author_v2.py +++ b/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: diff --git a/src/fi/alk/harness/call_runner.py b/src/fi/alk/harness/call_runner.py index e2ecb80d7..19fc972dc 100644 --- a/src/fi/alk/harness/call_runner.py +++ b/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) diff --git a/src/fi/alk/harness/certification.py b/src/fi/alk/harness/certification.py index 8631b9db1..ff2de9dcd 100644 --- a/src/fi/alk/harness/certification.py +++ b/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 diff --git a/src/fi/alk/harness/chat_call_runner.py b/src/fi/alk/harness/chat_call_runner.py index 7f2b629c3..c0c1297b3 100644 --- a/src/fi/alk/harness/chat_call_runner.py +++ b/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, diff --git a/src/fi/alk/harness/chat_readiness.py b/src/fi/alk/harness/chat_readiness.py index 6761d31ea..bd95d396d 100644 --- a/src/fi/alk/harness/chat_readiness.py +++ b/src/fi/alk/harness/chat_readiness.py @@ -1,8 +1,12 @@ """Private chat ingress and workflow readiness checks.""" import asyncio +import hashlib import json +import threading +import time from pathlib import Path +from urllib.parse import urlsplit from uuid import uuid4 from fi._diagnostics import _SECRET_FIELD, sanitize_diagnostic @@ -224,6 +228,457 @@ def scrub(value): return str(path) +TOOL_ROUTING = "tool-routing-observations.json" + + +def world_call_log(runtime, since=0.0): + """The calls the test world itself received since ``since``: its call log, written by + the world's own surfaces (the tool proxy in front of a tools service, the MCP toolbox), + never by the agent.""" + from .call_runner import _collect_tool_trace_calls + + try: + return [call for call in _collect_tool_trace_calls(runtime) if call.at >= since] + except Exception: # noqa: BLE001 - no log to read, nothing received + return [] + + +def _logged_tools(contract, runtime, bundle: Path | None = None) -> set[str]: + """Tools the world logs every call of: a service tool behind the tool proxy, and a tool + the bundle's MCP toolbox serves. Any other tool runs inside the agent, so the world holds + its data but no record of the call.""" + endpoints = getattr(runtime, "endpoints", None) or {} + logged: set[str] = set() + if "tool_proxy" in endpoints: + logged |= {e.tool for e in contract.tool_entrypoints if e.mode == "service"} + if "mcp_toolbox" in endpoints and bundle is not None: + try: + served = json.loads( + (bundle / "generated" / "mcp-toolbox" / "toolbox.json").read_text(encoding="utf-8") + ).get("tools") + except (OSError, ValueError, AttributeError): + served = None + logged |= set(served) if isinstance(served, dict) else set() + return logged + + +def _route(endpoint: str) -> str: + return "/" + urlsplit(endpoint).path.strip("/") + + +def _tool_names(contract): + """Which tools a call can be: by the full path the proxy logs with it, else by its name. + A name is a tool's own, or for a service tool the last segment of its path, which is all an + older proxy logged; a segment two tools' paths share names both, so neither is credited.""" + names = {tool.name: {tool.name} for tool in contract.tools} + paths: dict[str, set[str]] = {} + segments: dict[str, set[str]] = {} + for entry in contract.tool_entrypoints: + if entry.mode == "service" and entry.endpoint.strip(): + route = _route(entry.endpoint) + paths.setdefault(route, set()).add(entry.tool) + segments.setdefault(route.rsplit("/", 1)[-1], set()).add(entry.tool) + + def tools(call) -> set[str]: + endpoint = getattr(call, "endpoint", "") + if endpoint and _route(endpoint) in paths: + return paths[_route(endpoint)] + name = str(getattr(call, "name", "") or "") + return names.get(name) or segments.get(name, set()) + + return tools + + +def _outcome(call) -> str: + if getattr(call, "refused", False): + return "refused" + return "ok" if getattr(call, "ok", True) else "error" + + +def _routing(contract, runtime, *, received, reported, bundle=None): + """Routing read from the world's side only. A tool the world received is routed. A tool + the agent reports calling that the world logs but never received is not routed, unless a + received call could have been it or another tool. Any other reported tool runs inside the + agent, so its routing cannot be observed.""" + names = _tool_names(contract) + arrived: dict[str, list] = {} + ambiguous: set[str] = set() + for call in received: + candidates = names(call) + if len(candidates) == 1: + arrived.setdefault(next(iter(candidates)), []).append(call) + else: + ambiguous |= candidates + tools = [ + { + "tool": tool, + "routed": True, + "calls": len(calls), + "outcomes": sorted({_outcome(call) for call in calls}), + } + for tool, calls in arrived.items() + ] + unobserved = [] + logged = _logged_tools(contract, runtime, bundle) + claimed = { + tool for call in reported for tool in names(call) if len(names(call)) == 1 + } + for tool in sorted(claimed - set(arrived)): + if tool in ambiguous: + unobserved.append({"tool": tool, "reason": "ambiguous_world_record"}) + elif tool in logged: + tools.append({"tool": tool, "routed": False, "reason": "not_received"}) + else: + unobserved.append({"tool": tool, "reason": "no_world_record"}) + return tools, unobserved + + +def _write_routing(authoring: Path, body) -> None: + body["tools"] = sorted(body["tools"], key=lambda one: one["tool"]) + body["unobserved"] = sorted(body["unobserved"], key=lambda one: one["tool"]) + path = authoring / "generic-harness" / TOOL_ROUTING + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(body, indent=2, sort_keys=True) + "\n", encoding="utf-8") + + +def record_tool_routing( + authoring: Path, scenario_key: str, contract, runtime, *, received, reported, bundle=None +) -> None: + """Which tools the qualification conversation routed to the test world, beside the tool + certification: what readiness reads as observed routing, per tool. ``received`` is what + the world got (its call log and the calls the harness ran there); ``reported`` is every + call the conversation shows, including those only the agent says it made.""" + tools, unobserved = _routing( + contract, runtime, received=received, reported=reported, bundle=bundle + ) + _write_routing( + authoring, + { + "schema_version": 1, + "source": "world_call_log", + "scenario_key": scenario_key, + "tools": tools, + "unobserved": unobserved, + }, + ) + + +_REACH_FIELDS = ("dsn_env", "config_key", "host", "port", "database", "loader_module") + + +def _outside_world(contract) -> set[str]: + """Tools the contract says cannot be routed to the test world: no reachable + implementation, or a dependency the environment cannot stand up a copy of.""" + document = contract.model_dump(mode="json") if hasattr(contract, "model_dump") else {} + names = { + str(entry.get("tool") or "") + for entry in document.get("tool_entrypoints") or [] + if entry.get("mode") == "unreachable" + or (entry.get("mode") == "service" and not entry.get("service")) + } + for dependency in document.get("dependencies") or []: + reached = dependency.get("reached") or {} + if not (dependency.get("engine") and any(reached.get(f) for f in _REACH_FIELDS)): + names.update(str(tool) for tool in dependency.get("used_by") or []) + return names + + +_TYPE_EXAMPLES = {"int": 1, "integer": 1, "number": 1, "float": 1.0, "bool": True, "boolean": True} + + +def _tool_request(tool) -> str: + """One request for one tool, built from its contract: each argument takes the + contract's first example value, else a plain value of its type.""" + values = {} + for arg in tool.args: + listed = tool.arg_values.get(arg) + if isinstance(listed, list) and listed: + values[arg] = listed[0] + elif listed not in (None, [], ""): + values[arg] = listed + else: + values[arg] = _TYPE_EXAMPLES.get(str(tool.arg_types.get(arg, "")).lower(), "test") + shown = ", ".join(f"{arg}={value!r}" for arg, value in values.items()) + about = f" ({tool.description.strip()})" if tool.description.strip() else "" + return ( + f"Please use your {tool.name} tool{about} now" + + (f" with {shown}" if shown else "") + + ". This runs in a disposable test world: call the tool once and tell me what it returns." + ) + + +# The per-tool requests after qualification: one fresh message per tool not yet shown, +# bounded so a check with many tools stays well inside the platform's check deadline. +TOOL_REQUEST_LIMIT = 8 +TOOL_REQUEST_SECONDS = 300.0 + + +async def request_uncovered_tools( + *, + interface, + endpoint, + runtime, + contract, + authoring: Path, + bundle: Path, + source: Path | None, + secret_values, + candidate_hash: str, + local_auth_values=None, + timeout=120.0, + budget=TOOL_REQUEST_SECONDS, + limit=TOOL_REQUEST_LIMIT, +) -> None: + """Ask the agent once for each tool the qualification conversation did not show. + + At most ``limit`` requests within ``budget`` seconds in all, so a check stays well + inside its deadline; the tools left over are recorded as ``not_asked``. + + A tool the contract says has effects is never asked (``has_effects``), whether or not the + world logs its calls: a logged call is observable, not safe. A tool the contract says + leaves the world (no reachable implementation, or a dependency the environment cannot + copy) is not asked either (``reaches_outside``). Each request is a fresh one-message + conversation driven like a scenario's; at the world only the asked tool runs, and any + other call the agent returns is refused and listed under ``refused_during_requests``. A + tool whose contract leaves ``effects`` empty is asked: the asked tool runs only in the test +world, which is reset after the check, so an effect the contract did not name changes only +that world, never a real system (tools that leave the world are never asked). A callback agent + runs its own tools before answering, which the harness cannot stop; the world is reset + after the check. Routing is read from the world's side as in the conversation. Never + raises. + """ + from .call_runner import _collect_file_tool_calls + from .chat_call_runner import _HostedChatTarget, _tool_world + + path = authoring / "generic-harness" / TOOL_ROUTING + try: + body = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return + body.setdefault("unobserved", []) + shown = {one["tool"] for one in [*body["tools"], *body["unobserved"]]} + outside = _outside_world(contract) + # Requests already sent by an earlier attempt of this check: their count and time spent + # stay within one bound for the whole check, and an answer is reused only for the same + # candidate and tool definition. + ledger = _request_ledger(authoring) + deadline = time.monotonic() + max(0.0, budget - ledger["seconds"]) + for tool in getattr(contract, "tools", None) or []: + if tool.name in shown: + continue + if tool.name in outside: + # Asking could reach a real system: the contract says the tool leaves the world. + body["unobserved"].append({"tool": tool.name, "reason": "reaches_outside"}) + continue + if getattr(tool, "effects", None): + # Asking could change something; a synthetic request must only read. + body["unobserved"].append({"tool": tool.name, "reason": "has_effects"}) + continue + fingerprint = _tool_fingerprint(candidate_hash, contract, tool) + earlier = ledger["asked"].get(tool.name) + if isinstance(earlier, dict) and earlier.get("fingerprint") == fingerprint: + body["tools"].extend(earlier["tools"]) + body["unobserved"].extend(earlier["unobserved"]) + continue + left = deadline - time.monotonic() + if ledger["requests"] >= limit or left <= 0: + body["unobserved"].append({"tool": tool.name, "reason": "not_asked"}) + continue + ledger["requests"] += 1 + request_timeout = min(timeout, left) + since = time.time() + started = time.monotonic() + try: + traced = len(_collect_file_tool_calls(runtime)) + key = f"tool-request-{uuid4().hex}" + target = _HostedChatTarget( + wrapper=create_chat_wrapper( + interface, + endpoint, + timeout=request_timeout, + total_timeout=request_timeout, + secret_values=secret_values, + local_auth_values=local_auth_values, + ), + contract=contract, + world=_OnlyAsked( + _tool_world(bundle, contract, runtime, source), + tool.name, + deadline=time.monotonic() + request_timeout, + ), + scenario_key=key, + scenario_id=key, + ) + # One bound for the whole request: say() may call the agent up to eight times. + async with asyncio.timeout(request_timeout): + await target.say(_tool_request(tool)) + reported = [ + *target.world.calls, + *_collect_file_tool_calls(runtime)[traced:], + ] + except Exception: # noqa: BLE001 - an unanswered request leaves the tool unobserved + failed = [{"tool": tool.name, "reason": "request_failed"}] + body["unobserved"].extend(failed) + _note_request( + ledger, tool.name, fingerprint, [], failed, time.monotonic() - started + ) + continue + tools, unobserved = _routing( + contract, + runtime, + received=[*world_call_log(runtime, since), *target.received], + reported=[call for call in reported if call.name == tool.name], + bundle=bundle, + ) + asked = [ + {**one, "source": "tool_request"} + for one in tools + if one["tool"] == tool.name + ] + left_over = [one for one in unobserved if one["tool"] == tool.name] + if target.world.refused: + body.setdefault("refused_during_requests", []).append( + {"asked": tool.name, "refused": sorted(set(target.world.refused))} + ) + if not asked and not left_over: + left_over = [{"tool": tool.name, "reason": "not_called"}] + body["tools"].extend(asked) + body["unobserved"].extend(left_over) + _note_request( + ledger, tool.name, fingerprint, asked, left_over, time.monotonic() - started + ) + _write_routing(authoring, body) + _save_ledger(authoring, ledger) + + +class _OnlyAsked: + """The world as one tool request sees it: only the asked tool runs. + + Any other call the agent returns is refused, not run, and named in ``refused``, so a + request for a read-only tool can never make the world run a tool with effects. The asked + tool runs in a worker thread bounded by the request's ``deadline``: the world runs a call + synchronously, so a handler that blocks would otherwise hold the request past it. + """ + + def __init__(self, world, asked: str, *, deadline: float): + self._world = world + self._asked = asked + self._deadline = deadline + self.refused: list[str] = [] + + def __getattr__(self, name): + return getattr(self._world, name) + + def handle_tool_call(self, tool_call, **context): + from .environment import ToolExecutionResult + from .world.runtime import Call + + name = str( + tool_call.get("name") or (tool_call.get("function") or {}).get("name") or "" + ) + if name == self._asked: + outcome: dict = {} + + def run(): + try: + outcome["result"] = self._world.handle_tool_call( + tool_call, **context + ) + except BaseException as exc: # noqa: BLE001 - re-raised on the caller's side + outcome["error"] = exc + + worker = threading.Thread(target=run, daemon=True) + worker.start() + worker.join(max(0.0, self._deadline - time.monotonic())) + if worker.is_alive(): + # The thread cannot be stopped; its late result is discarded with this world. + raise TimeoutError( + f"{self._asked} outlasted the readiness request deadline" + ) + if "error" in outcome: + raise outcome["error"] + return outcome["result"] + error = f"only {self._asked} may run during this readiness request" + self._world.calls.append( + Call( + name=name, + arguments={}, + ok=False, + refused=True, + error=error, + at=time.time(), + ) + ) + self.refused.append(name) + return ToolExecutionResult( + tool_call_id=tool_call.get("id") or tool_call.get("tool_call_id"), + tool_name=name or "unknown", + content=error, + result=None, + success=False, + error=error, + state_updates={}, + ) + + +TOOL_REQUESTS = "tool-requests.json" + + +def _tool_fingerprint(candidate_hash: str, contract, tool) -> str: + """What an earlier answer about ``tool`` holds for: the candidate and the tool's definition.""" + definition = { + "tool": tool.model_dump(mode="json"), + "entrypoints": [ + entry.model_dump(mode="json") + for entry in contract.tool_entrypoints + if entry.tool == tool.name + ], + } + text = json.dumps([candidate_hash, definition], sort_keys=True, default=str) + return "sha256:" + hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _request_ledger(authoring: Path) -> dict: + """The tool requests earlier attempts of this check sent, and the time they took.""" + path = authoring / "generic-harness" / TOOL_REQUESTS + try: + stored = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + stored = {} + asked = stored.get("asked") if isinstance(stored, dict) else None + seconds = stored.get("seconds") if isinstance(stored, dict) else None + requests = stored.get("requests") if isinstance(stored, dict) else None + return { + "asked": asked if isinstance(asked, dict) else {}, + "requests": requests if isinstance(requests, int) else 0, + "seconds": float(seconds) if isinstance(seconds, (int, float)) else 0.0, + } + + +def _note_request( + ledger: dict, + tool: str, + fingerprint: str, + tools: list, + unobserved: list, + spent: float, +) -> None: + ledger["asked"][tool] = { + "fingerprint": fingerprint, + "tools": tools, + "unobserved": unobserved, + } + ledger["seconds"] += max(0.0, spent) + + +def _save_ledger(authoring: Path, ledger: dict) -> None: + path = authoring / "generic-harness" / TOOL_REQUESTS + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(ledger, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + + async def probe_chat_workflow( *, job, @@ -310,6 +765,7 @@ async def retain_interrupted_conversation(exc): source_directory=source, ), ) + since = time.time() async with asyncio.timeout(timeout): outcome = await runner.run(scenario, runtime) graded = graded_calls(outcome.calls) @@ -329,6 +785,15 @@ async def retain_interrupted_conversation(exc): ) if not graded: receipt["tool_evidence_status"] = "not_observed" + record_tool_routing( + authoring, + scenario.scenario_key, + runner._contract, + runtime, + received=[*world_call_log(runtime, since), *runner.received], + reported=outcome.calls, + bundle=bundle, + ) receipt["status"] = "passed" except RuntimeValidationError: raise diff --git a/src/fi/alk/harness/failure_record.py b/src/fi/alk/harness/failure_record.py new file mode 100644 index 000000000..6123e8573 --- /dev/null +++ b/src/fi/alk/harness/failure_record.py @@ -0,0 +1,130 @@ +"""The structured record of why a hosted check or build failed, written beside its work. + +It records, from the error itself, whose failure the harness believes it was: ``domain`` is +``harness`` (a bug or capacity limit of ours), ``world`` (a process the harness adds to the +generated world failed for a reason of ours) or ``agent`` (anything else, the agent's code and +services included). ``message`` is one redacted line of the harness's own text; the agent's +output, which follows it in the error, is never copied. It adds detail for a reader; the +platform still decides whose failure it was from the guest log's phase line. + +The file is diagnostic only. The tested agent runs under the same OS user and can write +``/work``, so nothing may treat ``domain`` as authoritative (for gating a run or deciding who +owns a failure) until the writer is isolated or the record is authenticated. + +Each failed validation replaces the record and a passing one removes it, so after the run it +names the failure that ended it: a certify that does not repair re-raises a generic error that +is never recorded. +""" + +from __future__ import annotations + +import json +import os +import re +from pathlib import Path +from typing import Any + +FAILURE_RECORD = Path(os.environ.get("ALK_FAILURE_RECORD", "/work/failure.json")) +SCHEMA_VERSION = 1 +_MESSAGE_CHARS = 300 +# The processes the harness adds to a generated world (bundle_author_v2): the managed +# world-db, the generated tool-proxy and mcp-toolbox. +_OWNED_PROCESSES = frozenset({"world-db", "tool-proxy", "mcp-toolbox"}) +# Codes raised with an infrastructure domain at every raise site, whatever the process. +_HARNESS_CODES = frozenset( + { + "store_statement_failed", + "resource_capacity_unknown", + "resource_capacity_unavailable", + } +) +# A seed that failed only for the harness's own fingerprint check. +_OWNED_SEED_DIAGNOSTICS = frozenset({"source_model_fingerprint_mismatch"}) +# Where the harness's text ends and the agent's output lines begin (process_runtime). +_AGENT_OUTPUT = "; last output:" +_TYPED = re.compile( + r"(?:ProcessRuntimeError: )?([a-z_]+)/([a-z_]+)(?: \(([^)]+)\))?: (.*)", re.S +) + + +def _typed(exc: BaseException) -> tuple[str, str, str | None, str] | None: + """``(stage, code, process, detail)`` of the process failure behind ``exc``.""" + from .process_runtime import ProcessRuntimeError + + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, ProcessRuntimeError): + detail = str(current).partition(": ")[2] + return current.stage, current.code, current.process, detail + current = current.__cause__ or current.__context__ + found = _TYPED.fullmatch(_own_text(exc)) + return found.groups() if found else None # type: ignore[return-value] + + +def _diagnostic_codes(exc: BaseException) -> set[str]: + """The codes of the structured diagnostics the failure carries, nearest first.""" + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + diagnostics = getattr(current, "diagnostics", None) or () + if diagnostics: + return {str(getattr(item, "code", "")) for item in diagnostics} + current = current.__cause__ or current.__context__ + return set() + + +def _own_text(exc: BaseException) -> str: + return str(exc).split(_AGENT_OUTPUT, 1)[0].strip() + + +def _domain( + phase: str, typed: tuple[str, str, str | None, str] | None, codes: set[str] +) -> str: + """The same rule the platform applied to the log, from the error itself.""" + if phase != "runtime" or typed is None: + return "agent" + _stage, code, process, _detail = typed + if code.startswith("internal_") or code in _HARNESS_CODES: + return "harness" + if process not in _OWNED_PROCESSES: + return "agent" + if code != "seed_failed": + return "world" + return "world" if codes and codes <= _OWNED_SEED_DIAGNOSTICS else "agent" + + +def record( + phase: str, exc: BaseException, path: Path | None = None +) -> dict[str, Any] | None: + """Write this failure in place of any earlier one. Never raises.""" + from .outbound import redact_outbound_text + + target = path or FAILURE_RECORD + try: + typed = _typed(exc) + line = " ".join(redact_outbound_text(_own_text(exc)).split()) + body = { + "schema_version": SCHEMA_VERSION, + "domain": _domain(phase, typed, _diagnostic_codes(exc)), + "phase": phase, + "stage": typed[0] if typed else None, + "code": typed[1] if typed else None, + "process": typed[2] if typed else None, + "message": line[:_MESSAGE_CHARS], + } + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(json.dumps(body, sort_keys=True) + "\n", encoding="utf-8") + return body + except Exception: # noqa: BLE001 - the record never changes how the run ends + return None + + +def clear(path: Path | None = None) -> None: + """A validation passed: no earlier failure ended the run. Never raises.""" + try: + (path or FAILURE_RECORD).unlink(missing_ok=True) + except OSError: + pass diff --git a/src/fi/alk/harness/generated_runtime.py b/src/fi/alk/harness/generated_runtime.py index 74b2e2924..deca04624 100644 --- a/src/fi/alk/harness/generated_runtime.py +++ b/src/fi/alk/harness/generated_runtime.py @@ -14,6 +14,7 @@ import shlex import shutil import stat +import subprocess from collections.abc import Sequence from dataclasses import asdict, dataclass from pathlib import Path @@ -687,6 +688,95 @@ def declared(body: dict[str, Any]) -> set[str]: # does, has them recorded only once the SDK is there, at the API's own version. ("opentelemetry-api", "opentelemetry-sdk"), ) +# Distributions that only mark how spans are made, not an agent framework of their own. +_NOT_FRAMEWORKS = frozenset({"opentelemetry-api"}) +_FRAMEWORKS_READ_SECONDS = 20 + + +def _distribution_name(name: str) -> str: + return re.sub(r"[-_.]+", "-", name).lower() + + +def installed_frameworks(source: Path | None = None) -> dict[str, str]: + """The agent framework distributions installed in the agent's environment, by version. + + Read from a virtual environment inside ``source`` when it has one, else from the ``python3`` the + generated runtime installs the agent into. That interpreter also holds the harness's own + dependencies, so from it only the frameworks the agent's dependency files name are counted. + Anything unreadable reads as none: the certificate then records no frameworks rather than wrong + ones. + """ + wanted = { + _distribution_name(name) + for name, _ in FRAMEWORK_INSTRUMENTORS + if name not in _NOT_FRAMEWORKS + } + found: dict[str, str] = {} + paths = ( + [ + str(path) + for pattern in (".venv/lib/python*/site-packages", "*/.venv/lib/python*/site-packages") + for path in sorted(source.glob(pattern)) + ] + if source is not None and source.is_dir() + else [] + ) + if paths: + import importlib.metadata + + for distribution in importlib.metadata.distributions(path=paths): + name = _distribution_name(distribution.metadata["Name"] or "") + if name in wanted: + found.setdefault(name, distribution.version) + return dict(sorted(found.items())) + wanted = _declared(source, wanted) + if not wanted: + return {} + script = ( + "import importlib.metadata as m, json, re\n" + "n = lambda s: re.sub(r'[-_.]+', '-', s).lower()\n" + f"w = {sorted(wanted)!r}\n" + "print(json.dumps({n(d.metadata['Name']): d.version for d in m.distributions()" + " if d.metadata['Name'] and n(d.metadata['Name']) in w}))\n" + ) + try: + completed = subprocess.run( + ["python3", "-c", script], + capture_output=True, + text=True, + timeout=_FRAMEWORKS_READ_SECONDS, + check=False, + ) + read = json.loads(completed.stdout or "{}") if completed.returncode == 0 else {} + except (OSError, ValueError, subprocess.TimeoutExpired): + return {} + if not isinstance(read, dict): + return {} + return { + str(name): str(version) + for name, version in sorted(read.items()) + if name in wanted and version + } + + +_DEPENDENCY_FILES = ("requirements*.txt", "pyproject.toml", "setup.py", "setup.cfg", "Pipfile") + + +def _declared(source: Path | None, wanted: set[str]) -> set[str]: + """The frameworks of ``wanted`` that the dependency files at the top of ``source`` name.""" + if source is None or not source.is_dir(): + return set() + text = "" + for pattern in _DEPENDENCY_FILES: + for path in sorted(source.glob(pattern)): + try: + text += path.read_text(encoding="utf-8", errors="replace")[:200_000].lower() + "\n" + except OSError: + continue + named = {_distribution_name(token) for token in re.findall(r"[a-z0-9][a-z0-9._-]*", text)} + return wanted & named + + # Half the build step's default limit, so a stalled index skips the instrumentors and never fails the build. INSTRUMENTOR_INSTALL_SECONDS = 300 diff --git a/src/fi/alk/harness/hosted_agent_reading.py b/src/fi/alk/harness/hosted_agent_reading.py new file mode 100644 index 000000000..29ac55889 --- /dev/null +++ b/src/fi/alk/harness/hosted_agent_reading.py @@ -0,0 +1,76 @@ +"""Read an agent version's tools inside a hosted check, for one the platform has never read. + +A pairing check certifies against the environment's saved contract: it talks to the agent, but +never reads a new agent version's own tools from its source. When the platform asks, this runs the +understand stage (a model call) on the uploaded source first, with the same generation credentials +an authoring run gets, +and writes the agent document the platform stores for the version: ``agent.json`` holding +``{"sdk_version", "agent"}``, built from the contract the way a build's registration builds it. + +The reading never decides the check: any failure is printed on one line and exits 0, so the +certify that follows runs as it would without it. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +from .. import __version__ as _sdk_version + +AGENT_DOCUMENT = "agent.json" + + +def _read(source: Path, output: Path, name: str) -> None: + from ..providers.agent import build_authored_agent + from . import cli, spend + from .contract import AgentContract + from .hosted_authoring_entrypoint import configure_authoring_credentials + + configure_authoring_credentials() + output.mkdir(parents=True, exist_ok=True) + journal = os.environ.get(spend.JOURNAL_ALIAS, "").strip() + if journal: + spend.journal_to(journal) + status = cli.main( + [ + "understand", + "--name", + name, + "--path", + str(source), + "--out", + str(output), + "--once", + ] + ) + if status != 0: + raise RuntimeError(f"understand exited {status}") + contract = AgentContract.model_validate_json( + (output / "contract.json").read_text(encoding="utf-8") + ).model_dump(mode="json") + document = {"sdk_version": _sdk_version, "agent": build_authored_agent(contract)} + (output / AGENT_DOCUMENT).write_text(json.dumps(document) + "\n", encoding="utf-8") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="hosted_agent_reading") + parser.add_argument("--source", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--name", default="agent") + args = parser.parse_args(argv) + try: + _read(args.source.resolve(), args.output.resolve(), args.name) + except ( + Exception + ) as exc: # noqa: BLE001 - a failed reading leaves the version unread + (args.output / AGENT_DOCUMENT).unlink(missing_ok=True) + print(f"agent reading failed: {type(exc).__name__}", file=sys.stderr) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/fi/alk/harness/hosted_authoring_entrypoint.py b/src/fi/alk/harness/hosted_authoring_entrypoint.py index 4ca795fe2..4ee903d09 100644 --- a/src/fi/alk/harness/hosted_authoring_entrypoint.py +++ b/src/fi/alk/harness/hosted_authoring_entrypoint.py @@ -146,7 +146,9 @@ def _authoring_job_context(forwarded: list[str]) -> tuple[str, str, dict]: return "", "", {} -def main(argv: list[str] | None = None) -> int: +def configure_authoring_credentials() -> dict[str, str]: + """Put the platform's generation credentials in this process's environment; return every + value the secrets channel holds.""" # This process authors the environment but never runs the submitted target. Do not let a # launcher-provided parent environment accidentally bypass the target-secrets file boundary. for name in _TARGET_CREDENTIAL_NAMES: @@ -178,6 +180,11 @@ def main(argv: list[str] | None = None) -> int: ) _configure_generation_environment(values) _configure_observability_environment(all_values) + return all_values + + +def main(argv: list[str] | None = None) -> int: + all_values = configure_authoring_credentials() target_values = { name: all_values[name] for name in _TARGET_CREDENTIAL_NAMES diff --git a/src/fi/alk/harness/judge.py b/src/fi/alk/harness/judge.py index a816bba56..8524eaf5f 100644 --- a/src/fi/alk/harness/judge.py +++ b/src/fi/alk/harness/judge.py @@ -148,6 +148,35 @@ def judge_model() -> str: return os.environ.get(JUDGE_MODEL_ALIAS, "").strip() or chosen_model() +_REACHABLE_TIMEOUT_SECONDS = 20 + + +def judge_reachable(timeout: float = _REACHABLE_TIMEOUT_SECONDS) -> str: + """``passed`` when the judge's model answers one tiny request in ``timeout`` seconds, else + ``failed``: a judged sub-goal cannot be graded without it.""" + try: + env = provider_env(judge_model()) + base, key = env.get("ANTHROPIC_BASE_URL"), env.get("ANTHROPIC_AUTH_TOKEN") + if not base or not key: + raise NoGateway("no gateway configured for the judge") + response = requests.post( + f"{base}/v1/chat/completions", + headers={"Authorization": f"Bearer {key}"}, + json={ + "model": env["ANTHROPIC_MODEL"], + "messages": [{"role": "user", "content": "Reply with the single word: ready"}], + "max_tokens": 5, + }, + timeout=timeout, + ) + response.raise_for_status() + answered = bool((response.json().get("choices") or [None])[0]) + except Exception as exc: # noqa: BLE001 - any failure means the judge cannot grade + logger.warning("judge not reachable in the check: %s", type(exc).__name__) + return "failed" + return "passed" if answered else "failed" + + def _short(value: object) -> object: if isinstance(value, str) and len(value) > _LIMIT: return value[:_LIMIT] + f"... [{len(value)} chars]" diff --git a/src/fi/alk/harness/mcp_toolbox_adapter.py b/src/fi/alk/harness/mcp_toolbox_adapter.py index 7650a04be..851a4abe3 100644 --- a/src/fi/alk/harness/mcp_toolbox_adapter.py +++ b/src/fi/alk/harness/mcp_toolbox_adapter.py @@ -69,6 +69,23 @@ def _execute( error = f"{type(exc).__name__}: {exc}" raise finally: + try: + # The world's own call log, as the tool proxy keeps it: what reached the world. + with psycopg.connect(database_url, autocommit=True) as log: + log.execute( + "INSERT INTO _alk_tool_trace(name, arguments, result, ok, error, at) " + "VALUES (%s, %s, %s, %s, %s, %s)", + ( + tool_name, + json.dumps(arguments, default=str), + json.dumps(result, default=str), + error is None, + error or "", + started, + ), + ) + except Exception: # noqa: BLE001 - the log must never change the tool's response + pass trace_path.parent.mkdir(parents=True, exist_ok=True) with trace_path.open("a", encoding="utf-8") as trace: trace.write( @@ -150,7 +167,20 @@ async def toolset_compatible( if scope.get("type") == "http" and str(scope.get("path") or "").startswith( "/mcp/" ): - scope = {**scope, "path": "/mcp/", "raw_path": b"/mcp/"} + # Toolbox clients send ``Accept: */*``; the JSON-response server requires the + # JSON type to be named, so a bare wildcard would refuse every client. + headers = [ + (name, value) + for name, value in scope.get("headers") or [] + if name.lower() != b"accept" + ] + accept = b", ".join( + value for name, value in scope.get("headers") or [] if name.lower() == b"accept" + ) + if b"application/json" not in accept: + accept = b", ".join(filter(None, (b"application/json", accept))) + headers.append((b"accept", accept)) + scope = {**scope, "path": "/mcp/", "raw_path": b"/mcp/", "headers": headers} await application(scope, receive, send) return toolset_compatible diff --git a/src/fi/alk/harness/process_runtime.py b/src/fi/alk/harness/process_runtime.py index a30d49983..1756850a9 100644 --- a/src/fi/alk/harness/process_runtime.py +++ b/src/fi/alk/harness/process_runtime.py @@ -46,7 +46,7 @@ from datetime import datetime, timezone from enum import Enum from pathlib import Path -from typing import Any, Callable, Protocol, Sequence +from typing import TYPE_CHECKING, Any, Callable, Protocol, Sequence from urllib.parse import quote as urlquote, urlsplit from pydantic import BaseModel, Field, JsonValue, PrivateAttr, ValidationError @@ -86,6 +86,9 @@ validate_provision_receipt, ) +if TYPE_CHECKING: + from .source_model import SourceModel + logger = logging.getLogger(__name__) # --- errors -------------------------------------------------------------------------------- @@ -3584,6 +3587,37 @@ def apply_postgres_sqlite_world( artifacts.write_world_ir(imported.world) +def _without_later_harness_tables( + live: SourceModel, recorded: SourceModel +) -> SourceModel: + """The live schema with the harness's own `_alk_` tables as the build recorded them, or left out where it had + none: they are not the agent's data, and another harness version adds or widens them.""" + from .source_model import SourceModel + + built = {table.name: table for table in recorded.tables} + tables = tuple( + built.get(table.name, table) if table.name.startswith("_alk_") else table + for table in live.tables + if not table.name.startswith("_alk_") or table.name in built + ) + if tables == live.tables: + return live + return SourceModel.create( + source_digest=live.source_digest, + engine=live.engine, + engine_version=live.engine_version, + tables=tables, + processes=live.processes, + interfaces=live.interfaces, + actions=live.actions, + configuration_names=live.configuration_names, + migrations=live.migrations, + seeds=live.seeds, + evidence=live.evidence, + unsupported=live.unsupported, + ) + + def _compose_seed_source_model(source, seed_file: Path, artifact_root: Path | None): """Reconcile bundled code facts with the independently inspected live schema.""" from .certification import GenericHarnessArtifactStore @@ -3601,8 +3635,9 @@ def _compose_seed_source_model(source, seed_file: Path, artifact_root: Path | No raise ValueError("seed_source_model_symlink") if path.is_file(): try: + recorded = GenericHarnessArtifactStore(root).read_source_model() return compose_source_models( - GenericHarnessArtifactStore(root).read_source_model(), source + recorded, _without_later_harness_tables(source, recorded) ) except ValueError as error: # Code facts recorded for other source than the schema running here. diff --git a/src/fi/alk/harness/tool_trace_proxy.py b/src/fi/alk/harness/tool_trace_proxy.py index 061a97cb2..c187caadf 100644 --- a/src/fi/alk/harness/tool_trace_proxy.py +++ b/src/fi/alk/harness/tool_trace_proxy.py @@ -16,15 +16,21 @@ def _record( - name: str, arguments: object, result: object, ok: bool, failure: str = "" + name: str, + endpoint: str, + arguments: object, + result: object, + ok: bool, + failure: str = "", ) -> None: try: with psycopg.connect(DATABASE_URL, autocommit=True) as connection: connection.execute( - "INSERT INTO _alk_tool_trace(name, arguments, result, ok, error, at) " - "VALUES (%s, %s, %s, %s, %s, %s)", + "INSERT INTO _alk_tool_trace(name, endpoint, arguments, result, ok, error, at) " + "VALUES (%s, %s, %s, %s, %s, %s, %s)", ( name, + endpoint, json.dumps(arguments), json.dumps(result), ok, @@ -66,7 +72,8 @@ def _forward(self) -> None: "content-type": self.headers.get("content-type", "application/json") }, ) - name = self.path.split("?", 1)[0].rstrip("/").rsplit("/", 1)[-1] or "unknown" + endpoint = self.path.split("?", 1)[0] + name = endpoint.rstrip("/").rsplit("/", 1)[-1] or "unknown" try: with request.urlopen(outgoing, timeout=30) as response: content = response.read() @@ -76,7 +83,7 @@ def _forward(self) -> None: result = json.loads(content) if content else None except ValueError: result = content.decode("utf-8", errors="replace") - _record(name, arguments, result, status < 400) + _record(name, endpoint, arguments, result, status < 400) self.send_response(status) self.send_header("content-type", response_type) self.send_header("content-length", str(len(content))) @@ -85,13 +92,20 @@ def _forward(self) -> None: except error.HTTPError as exc: content = exc.read() failure = content.decode("utf-8", errors="replace")[:2000] - _record(name, arguments, None, False, failure) + _record(name, endpoint, arguments, None, False, failure) self.send_response(exc.code) self.send_header("content-length", str(len(content))) self.end_headers() self.wfile.write(content) except Exception as exc: - _record(name, arguments, None, False, f"{type(exc).__name__}: unavailable") + _record( + name, + endpoint, + arguments, + None, + False, + f"{type(exc).__name__}: unavailable", + ) content = json.dumps({"detail": "tool_upstream_unavailable"}).encode() self.send_response(502) self.send_header("content-type", "application/json") diff --git a/src/fi/alk/harness/world/runtime.py b/src/fi/alk/harness/world/runtime.py index 7ae12aa5d..71defa4df 100644 --- a/src/fi/alk/harness/world/runtime.py +++ b/src/fi/alk/harness/world/runtime.py @@ -172,6 +172,8 @@ class Call: # Part of the arguments or the result was left out to keep the record bounded. input_omitted: bool = False output_omitted: bool = False + # The HTTP path a call through the tool proxy went to; empty for any other source. + endpoint: str = "" def tool_call_attribution(record: Mapping[str, Any]) -> dict[str, Any]: diff --git a/tests/harness/test_authoring_runtime_validation.py b/tests/harness/test_authoring_runtime_validation.py index 3e37bf63d..82729a13f 100644 --- a/tests/harness/test_authoring_runtime_validation.py +++ b/tests/harness/test_authoring_runtime_validation.py @@ -1076,7 +1076,11 @@ def write_action_certification(self, _report): def test_generic_certificate_combines_runtime_evidence_and_repair_history( tmp_path, source_invariants, + monkeypatch, ) -> None: + import fi.alk + + monkeypatch.setattr(fi.alk, "__version__", "0.3.1") authoring = tmp_path / "authoring" store = GenericHarnessArtifactStore(authoring / "generic-harness") @@ -1110,11 +1114,21 @@ def digest(character): metadata={"daytona_snapshot": "snapshot-r1"}, ) - _write_generic_certificate(job, authoring, controller.history) + agent = tmp_path / "agent" + info = agent / ".venv/lib/python3.11/site-packages/langchain_core-0.3.12.dist-info" + info.mkdir(parents=True) + (info / "METADATA").write_text( + "Metadata-Version: 2.1\nName: langchain-core\nVersion: 0.3.12\n", encoding="utf-8" + ) + + _write_generic_certificate(job, authoring, controller.history, agent) certificate = store.read_certification() assert certificate.status.value == "certified" assert certificate.source.repository == "future-agi/example" + # The frameworks are read from the agent's own source, beside the job's repository. + assert certificate.runtime.frameworks == {"langchain-core": "0.3.12"} + assert certificate.runtime.harness_version == "0.3.1" assert certificate.runtime.snapshot == "snapshot-r1" assert certificate.runtime.validation_attempts == 1 assert certificate.checks.scenario_setup_ready == "2/2" diff --git a/tests/harness/test_call_runner.py b/tests/harness/test_call_runner.py index 645c4fb42..18b31afd4 100644 --- a/tests/harness/test_call_runner.py +++ b/tests/harness/test_call_runner.py @@ -1831,6 +1831,15 @@ def test_tool_trace_translates_rows_and_applies_v1_refused_rule(monkeypatch) -> ) +def test_tool_trace_keeps_the_path_the_tool_proxy_logged(monkeypatch) -> None: + columns = ["id", "name", "endpoint", "arguments", "result", "ok", "error", "at"] + rows = [(1, "search", "/orders/search", {}, None, True, "", 1.0)] + monkeypatch.setitem(sys.modules, "psycopg", _FakePsycopg(rows, columns)) + runtime = _runtime(endpoints={"database": _postgres_endpoint()}) + + assert cr._collect_tool_trace_calls(runtime)[0].endpoint == "/orders/search" + + def test_tool_trace_read_failure_degrades_to_no_calls_never_crashes( monkeypatch, ) -> None: diff --git a/tests/harness/test_certification.py b/tests/harness/test_certification.py index cdbe5d4f6..d42c46163 100644 --- a/tests/harness/test_certification.py +++ b/tests/harness/test_certification.py @@ -443,3 +443,134 @@ def test_chat_applicability_uses_bound_contract_and_provider_mode( agent=SimpleNamespace(mode=ProviderExecutionMode.CONNECT_ONLY), ) assert requires_chat_ingress(job, tmp_path) is required + + +def test_old_certificate_keeps_its_fingerprint() -> None: + certificate = _certificate() + body = certificate.model_dump_json() + + for unrecorded in ("harness_version", "frameworks", "judge_reachable"): + assert unrecorded not in body + again = HarnessCertification.model_validate_json(body) + assert again.fingerprint == certificate.fingerprint + assert again.model_dump_json() == body + + +def test_certificate_records_harness_and_frameworks() -> None: + held = _certificate() + certificate = HarnessCertification.create( + status=held.status, + source=held.source, + authoring=held.authoring, + compiler=held.compiler, + runtime=CertificationRuntime( + snapshot="snapshot-r1", + validation_attempts=1, + harness_version="0.3.1", + frameworks={"langchain-core": "0.3.12"}, + ), + checks=held.checks.model_copy(update={"judge_reachable": CheckStatus.FAILED}), + repairs=held.repairs, + limitations=held.limitations, + ) + body = json.loads(certificate.model_dump_json()) + + assert body["runtime"]["harness_version"] == "0.3.1" + assert body["runtime"]["frameworks"] == {"langchain-core": "0.3.12"} + assert body["checks"]["judge_reachable"] == "failed" + assert certificate.fingerprint != held.fingerprint + assert HarnessCertification.model_validate(body).fingerprint == certificate.fingerprint + + +def _authoring_with(tmp_path: Path, *, judged: str) -> Path: + authoring = tmp_path / "authoring" + authoring.mkdir() + (authoring / "sub_goals.json").write_text( + json.dumps( + { + "sub_goals": [ + {"name": "refund_issued", "check": "def check(world, calls):\n return None\n"}, + {"name": "refusal_explained", "judged": judged}, + ] + } + ), + encoding="utf-8", + ) + return authoring + + +@pytest.mark.parametrize(("judged", "calls"), [("Pass: the refusal names the policy.", 1), ("", 0)]) +def test_judge_check_runs_only_with_judged_sub_goals( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, judged: str, calls: int +) -> None: + from fi.alk.harness import authoring_runtime_validation, judge + + asked: list[float] = [] + + def unreachable(timeout: float = 20) -> str: + asked.append(timeout) + return "failed" + + monkeypatch.setattr(judge, "judge_reachable", unreachable) + checks = _certificate().checks + + found = authoring_runtime_validation._with_judge_check( + checks, _authoring_with(tmp_path, judged=judged) + ) + + assert len(asked) == calls + assert found.judge_reachable == (CheckStatus.FAILED if calls else None) + + +def test_judge_unreachable_without_a_gateway(monkeypatch: pytest.MonkeyPatch) -> None: + from fi.alk.harness import judge + + monkeypatch.setattr(judge, "provider_env", lambda model: {}) + + assert judge.judge_reachable(timeout=1) == "failed" + + +def test_frameworks_are_read_from_the_agents_environment(tmp_path: Path) -> None: + from fi.alk.harness.generated_runtime import installed_frameworks + + packages = tmp_path / ".venv" / "lib" / "python3.11" / "site-packages" + for name, version in ( + ("langchain_core", "0.3.12"), + ("opentelemetry_api", "1.30.0"), + ("requests", "2.32.0"), + ): + info = packages / f"{name}-{version}.dist-info" + info.mkdir(parents=True) + (info / "METADATA").write_text( + f"Metadata-Version: 2.1\nName: {name.replace('_', '-')}\nVersion: {version}\n", + encoding="utf-8", + ) + + # A distribution that only marks how spans are made is not a framework. + assert installed_frameworks(tmp_path) == {"langchain-core": "0.3.12"} + + +def test_the_interpreter_counts_only_the_frameworks_the_agent_declares( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + import json + import subprocess + + from fi.alk.harness import generated_runtime + + # The interpreter the agent is installed into also holds the harness's own dependencies. + installed = {"claude-agent-sdk": "0.2.156", "google-adk": "2.7.1", "crewai": "0.80.0"} + + def run(command, **_kwargs): + listed = json.loads(command[2].split("w = ", 1)[1].split("\n", 1)[0].replace("'", '"')) + found = {name: version for name, version in installed.items() if name in listed} + return subprocess.CompletedProcess(command, 0, stdout=json.dumps(found), stderr="") + + monkeypatch.setattr(generated_runtime.subprocess, "run", run) + (tmp_path / "requirements.txt").write_text("crewai==0.80.0\nrequests\n", encoding="utf-8") + + assert generated_runtime.installed_frameworks(tmp_path) == {"crewai": "0.80.0"} + # An agent that declares none records none, not the harness's own. + (tmp_path / "requirements.txt").write_text("requests\n", encoding="utf-8") + assert generated_runtime.installed_frameworks(tmp_path) == {} + assert generated_runtime.installed_frameworks(None) == {} diff --git a/tests/harness/test_chat_readiness.py b/tests/harness/test_chat_readiness.py index 77651c357..6e7cd242e 100644 --- a/tests/harness/test_chat_readiness.py +++ b/tests/harness/test_chat_readiness.py @@ -893,6 +893,53 @@ def test_private_tool_artifacts_preserve_json_lines_during_redaction(tmp_path, c assert Path(path).stat().st_mode & 0o777 == 0o600 +def test_private_workflow_counts_a_returned_call_the_harness_ran_at_the_world( + tmp_path, profile_app, monkeypatch +): + from fi.alk.harness import chat_call_runner + from fi.alk.harness.run.conversation import Exchange, Transcript + + url, state = profile_app + state.update( + profile=True, + body={"content": "", "tool_calls": [{"id": "t-1", "name": "lookup_account", "arguments": {}}]}, + ) + + async def conversation(target, *_args): + async def answered(*_a, **_k): + response = await sent(*_a, **_k) + state["body"] = {"content": "Done"} # once the harness has run the call + return response + + sent = target._wrapper.call + target._wrapper.call = answered + reply = await target.say("Look up my account") + return Transcript( + exchanges=[Exchange("customer", "Look up my account"), Exchange("agent", reply)], + calls=list(target.world.calls), ended="finished", + ) + + class World: + calls: list = [] + + def handle_tool_call(self, call): + from fi.alk.harness.world.runtime import Call + + self.calls.append(Call(name=call["name"], arguments={})) + return SimpleNamespace(content="{}") + + monkeypatch.setattr(chat_call_runner, "_drive_conversation", conversation) + monkeypatch.setattr(chat_call_runner, "_tool_world", lambda *_a: World()) + contract = AgentContract( + runtime=Runtime(language="python", command=["python", "app.py"], interface=interface()), + tools=[{"name": "lookup_account", "args": []}], + ) + asyncio.run(workflow_probe(tmp_path, url, contract=contract)) + + routing = json.loads((tmp_path / "generic-harness" / "tool-routing-observations.json").read_text()) + assert [(one["tool"], one["routed"]) for one in routing["tools"]] == [("lookup_account", True)] + + @pytest.mark.parametrize("kind", ["trace", "transcript"]) @pytest.mark.parametrize("redactions", [None, {"credential": "x" * (1024 * 1024 + 1)}]) def test_private_json_artifacts_fail_closed_without_safe_credential_context(tmp_path, kind, redactions): @@ -958,6 +1005,12 @@ async def conversation(target, *_args): assert receipt["status"] == ("passed" if completed else "failed") assert receipt["tool_call_count"] == int(completed) assert receipt["candidate_hash"] == "candidate" + routing = tmp_path / "generic-harness" / "tool-routing-observations.json" + if completed: + # The callback ran the tool itself and the world received nothing: never routed. + assert json.loads(routing.read_text())["tools"] == [] + else: + assert not routing.exists() assert receipt["bundle_digest"] == "bundle" assert receipt["scenario_key"] == "private" assert receipt["turns"] == 4 diff --git a/tests/harness/test_failure_record.py b/tests/harness/test_failure_record.py new file mode 100644 index 000000000..b844c71d3 --- /dev/null +++ b/tests/harness/test_failure_record.py @@ -0,0 +1,255 @@ +from __future__ import annotations + +import json + +import pytest + +from fi.alk.harness import failure_record +from fi.alk.harness.authoring_runtime_validation import RuntimeValidationError +from fi.alk.harness.diagnostics import HarnessDiagnostic +from fi.alk.harness.job import HarnessStage +from fi.alk.harness.process_runtime import ProcessRuntimeError + +CANONICAL = "canonical World IR could not be compiled or applied: " + + +def _diagnostics(*codes): + return tuple( + HarnessDiagnostic.create( + stage=HarnessStage.VALIDATING_ENVIRONMENT, + component="world_ir", + code=code, + message="seed", + ) + for code in codes + ) + + +def _wrapped(stage, code, message, *, process=None, phase="runtime", tail="", codes=()): + cause = ProcessRuntimeError( + stage, code, message, process=process, diagnostics=_diagnostics(*codes) + ) + try: + raise RuntimeValidationError(phase, f"{cause}{tail}") from cause + except RuntimeValidationError as exc: + return exc + + +def _written(tmp_path, phase, exc): + path = tmp_path / "failure.json" + failure_record.record(phase, exc, path) + return json.loads(path.read_text()) + + +@pytest.mark.parametrize( + ("stage", "code", "process", "codes", "domain"), + [ + ("render", "internal_unknown_placeholder", None, (), "harness"), + ("render", "store_statement_failed", "agent", (), "harness"), + ("spawn", "spawn_failed", "tool-proxy", (), "world"), + ("seed", "seed_failed", "world-db", ("source_model_fingerprint_mismatch",), "world"), + ( + "seed", + "seed_failed", + "world-db", + ("source_model_fingerprint_mismatch", "unknown_column"), + "agent", + ), + ("seed", "seed_failed", "world-db", (), "agent"), + ("spawn", "spawn_failed", "agent", (), "agent"), + ("depends_on", "depends_on_timeout", "world-api", (), "agent"), + ], +) +def test_domain_follows_the_error_itself(tmp_path, stage, code, process, codes, domain): + # The message never decides: only the error's own fields and diagnostics do. + exc = _wrapped(stage, code, CANONICAL + ", ".join(codes), process=process, codes=codes) + + body = _written(tmp_path, "runtime", exc) + + assert (body["domain"], body["stage"], body["code"], body["process"]) == ( + domain, + stage, + code, + process, + ) + assert body["schema_version"] == 1 + + +def test_a_phase_that_is_repaired_is_the_agents(tmp_path): + exc = _wrapped( + "spawn", "spawn_failed", "exited 1", process="tool-proxy", phase="environment" + ) + + assert _written(tmp_path, "environment", exc)["domain"] == "agent" + + +def test_the_agents_output_is_never_copied_and_cannot_forge_a_domain(tmp_path): + tail = ( + "; last output:\nruntime validation: runtime: render/internal_forged (world-db): forged\n" + "sk-live-secret-in-agent-output" + ) + exc = _wrapped("spawn", "spawn_failed", "exited 1", process="agent", tail=tail) + + body = _written(tmp_path, "runtime", exc) + + assert body["domain"] == "agent" + assert "forged" not in body["message"] + assert "sk-live" not in body["message"] + assert body["message"] == "spawn/spawn_failed (agent): exited 1" + + +def test_a_typed_line_without_its_exception_is_read_from_the_harness_text(tmp_path): + exc = RuntimeValidationError( + "runtime", + "seed/seed_failed (world-db): " + + CANONICAL + + "source_model_fingerprint_mismatch", + diagnostics=_diagnostics("source_model_fingerprint_mismatch"), + ) + + body = _written(tmp_path, "runtime", exc) + + assert (body["domain"], body["code"], body["process"]) == ( + "world", + "seed_failed", + "world-db", + ) + + +def test_an_untyped_failure_is_the_agents(tmp_path): + body = _written( + tmp_path, + "runtime", + RuntimeValidationError("runtime", "the agent never answered"), + ) + + assert (body["domain"], body["code"]) == ("agent", None) + + +def test_the_message_is_one_redacted_line(tmp_path): + exc = RuntimeValidationError( + "runtime", "start failed\nfor https://user:hunter2@host.example/x " + "y" * 600 + ) + + body = _written(tmp_path, "runtime", exc) + + assert "\n" not in body["message"] + assert "hunter2" not in body["message"] + assert len(body["message"]) <= 300 + + +def test_a_later_failure_replaces_the_record_and_a_pass_clears_it(tmp_path): + path = tmp_path / "failure.json" + failure_record.record( + "runtime", _wrapped("spawn", "spawn_failed", "a", process="agent"), path + ) + failure_record.record( + "runtime", _wrapped("spawn", "spawn_failed", "b", process="tool-proxy"), path + ) + + assert json.loads(path.read_text())["domain"] == "world" + + failure_record.clear(path) + + assert not path.exists() + + +def test_writing_never_raises(tmp_path): + blocked = tmp_path / "file" + blocked.write_text("") + + assert ( + failure_record.record( + "runtime", RuntimeValidationError("runtime", "x"), blocked / "failure.json" + ) + is None + ) + + +@pytest.mark.parametrize("generic", [False, True]) +def test_a_validation_writes_the_failure_that_ended_it_and_a_pass_clears_it( + tmp_path, monkeypatch, generic +): + import asyncio + + from fi.alk.harness.authoring_runtime_validation import validate_and_repair + from tests.harness.test_workflow_intent import _job as _plain_job + + def _job(source): + job = _plain_job(source) + if generic: + job = job.model_copy( + update={"metadata": {**job.metadata, "generic_harness_v1": True}} + ) + return job + + path = tmp_path / "failure.json" + monkeypatch.setattr(failure_record, "FAILURE_RECORD", path) + attempts = 0 + + async def validate(*args): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise _wrapped("spawn", "spawn_failed", "exited 1", process="tool-proxy") + return 2 + + async def refuse(phase, guidance): + raise RuntimeValidationError(phase, "Certify-only does not repair") + + with pytest.raises(RuntimeValidationError): + asyncio.run( + validate_and_repair( + _job(tmp_path), tmp_path, tmp_path, validate=validate, repair=refuse + ) + ) + + body = json.loads(path.read_text()) + assert (body["domain"], body["message"]) == ( + "world", + "spawn/spawn_failed (tool-proxy): exited 1", + ) + + async def passes(*args): + return 2 + + asyncio.run( + validate_and_repair( + _job(tmp_path), tmp_path, tmp_path, validate=passes, repair=refuse + ) + ) + + assert not path.exists() + + +@pytest.mark.parametrize("generic", [False, True]) +def test_a_record_from_an_earlier_attempt_never_speaks_for_a_later_failure( + tmp_path, monkeypatch, generic +): + import asyncio + + from fi.alk.harness.authoring_runtime_validation import validate_and_repair + from tests.harness.test_workflow_intent import _job as _plain_job + + job = _plain_job(tmp_path) + if generic: + job = job.model_copy( + update={"metadata": {**job.metadata, "generic_harness_v1": True}} + ) + path = tmp_path / "failure.json" + monkeypatch.setattr(failure_record, "FAILURE_RECORD", path) + # What an earlier attempt that failed on our side left behind. + path.write_text(json.dumps({"domain": "world", "message": "seed/seed_failed (world-db)"})) + + async def crashes(*args): + raise RuntimeError("the agent crashed") + + async def refuse(phase, guidance): + raise RuntimeValidationError(phase, "Certify-only does not repair") + + with pytest.raises(RuntimeError): + asyncio.run( + validate_and_repair(job, tmp_path, tmp_path, validate=crashes, repair=refuse) + ) + + assert not path.exists() diff --git a/tests/harness/test_hosted_agent_reading.py b/tests/harness/test_hosted_agent_reading.py new file mode 100644 index 000000000..50f6061a2 --- /dev/null +++ b/tests/harness/test_hosted_agent_reading.py @@ -0,0 +1,144 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from fi.alk.harness import cli, hosted_agent_reading, hosted_authoring_entrypoint, spend +from fi.alk.harness.contract import AgentContract + + +def _contract() -> str: + return AgentContract.model_validate( + { + "agent": "shop-agent", + "modality": "chat", + "tools": [{"name": "send_email", "args": ["to", "body"]}], + } + ).model_dump_json() + + +@pytest.fixture +def reading(monkeypatch, tmp_path): + calls: list[list[str]] = [] + monkeypatch.setattr( + hosted_authoring_entrypoint, "configure_authoring_credentials", lambda: {} + ) + + def understand(argv, contract=_contract(), status=0): + calls.append(argv) + out = Path(argv[argv.index("--out") + 1]) + if contract is not None: + (out / "contract.json").write_text(contract, encoding="utf-8") + return status + + monkeypatch.setattr(cli, "main", understand) + return calls, tmp_path / "reading" + + +def test_reading_writes_the_agent_document_from_the_contract(reading, tmp_path): + calls, output = reading + + status = hosted_agent_reading.main( + ["--source", str(tmp_path), "--output", str(output), "--name", "shop"] + ) + + assert status == 0 + assert calls[0][:3] == ["understand", "--name", "shop"] + assert "--once" in calls[0] + document = json.loads((output / "agent.json").read_text()) + assert document["sdk_version"] + assert [tool["name"] for tool in document["agent"]["tools"]] == ["send_email"] + assert set(document["agent"]["tools"][0]["parameters"]["properties"]) == { + "to", + "body", + } + assert document["agent"]["kind"] == "source" + + +def test_a_failed_understand_leaves_no_document_and_exits_zero( + monkeypatch, reading, tmp_path, capsys +): + _, output = reading + monkeypatch.setattr(cli, "main", lambda argv: 1) + + status = hosted_agent_reading.main( + ["--source", str(tmp_path), "--output", str(output)] + ) + + assert status == 0 + assert not (output / "agent.json").exists() + assert "agent reading failed" in capsys.readouterr().err + + +def test_an_unreadable_contract_leaves_no_document(monkeypatch, reading, tmp_path): + _, output = reading + + def broken(argv): + out = Path(argv[argv.index("--out") + 1]) + (out / "contract.json").write_text("{not json", encoding="utf-8") + return 0 + + monkeypatch.setattr(cli, "main", broken) + + assert ( + hosted_agent_reading.main(["--source", str(tmp_path), "--output", str(output)]) + == 0 + ) + assert not (output / "agent.json").exists() + + +def test_the_reading_spend_goes_to_the_named_journal(monkeypatch, reading, tmp_path): + _, output = reading + journal = tmp_path / "cost.json" + monkeypatch.setenv(spend.JOURNAL_ALIAS, str(journal)) + seen: list[object] = [] + monkeypatch.setattr(spend, "journal_to", lambda path: seen.append(path)) + + hosted_agent_reading.main(["--source", str(tmp_path), "--output", str(output)]) + + assert seen == [str(journal)] + + +def test_credentials_are_configured_before_understand(monkeypatch, reading, tmp_path): + _, output = reading + order: list[str] = [] + monkeypatch.setattr( + hosted_authoring_entrypoint, + "configure_authoring_credentials", + lambda: order.append("credentials") or {}, + ) + monkeypatch.setattr(cli, "main", lambda argv: order.append("understand") or 1) + + hosted_agent_reading.main(["--source", str(tmp_path), "--output", str(output)]) + + assert order == ["credentials", "understand"] + + +def test_a_failed_understand_with_a_partial_contract_writes_nothing( + monkeypatch, reading, tmp_path +): + _, output = reading + + def partial(argv): + out = Path(argv[argv.index("--out") + 1]) + (out / "contract.json").write_text(_contract(), encoding="utf-8") + return 1 + + monkeypatch.setattr(cli, "main", partial) + + hosted_agent_reading.main(["--source", str(tmp_path), "--output", str(output)]) + + assert not (output / "agent.json").exists() + + +def test_a_failed_reading_removes_an_earlier_document(monkeypatch, reading, tmp_path): + _, output = reading + output.mkdir() + (output / "agent.json").write_text("{}", encoding="utf-8") + monkeypatch.setattr(cli, "main", lambda argv: 1) + + hosted_agent_reading.main(["--source", str(tmp_path), "--output", str(output)]) + + assert not (output / "agent.json").exists() diff --git a/tests/harness/test_mcp_toolbox_adapter.py b/tests/harness/test_mcp_toolbox_adapter.py index 4856c681f..e8ca4dbd8 100644 --- a/tests/harness/test_mcp_toolbox_adapter.py +++ b/tests/harness/test_mcp_toolbox_adapter.py @@ -121,7 +121,11 @@ def execute(self, statement, parameters): database_url="unused", trace_path=tmp_path / "trace.jsonl", ) - assert captured == [(statement.encode(), [b"one", b"two"])] + assert captured[0] == (statement.encode(), [b"one", b"two"]) + # Then the call goes into the world's own call log, as the tool proxy logs its calls. + logged, values = captured[1] + assert logged.startswith(b"INSERT INTO _alk_tool_trace(name, arguments, result, ok, error, at)") + assert values[:2] == [b"example", b'{"first": "one", "second": "two"}'] def test_source_sql_executes_without_rewriting_literals_or_comments(tmp_path): @@ -153,3 +157,37 @@ def test_source_sql_executes_without_rewriting_literals_or_comments(tmp_path): } ] assert json.loads(trace.read_text())["is_error"] is False + + +@pytest.mark.parametrize("accept", [b"*/*", None]) +def test_a_toolbox_client_that_does_not_name_json_is_answered(tmp_path, accept): + from starlette.testclient import TestClient + + application = _application( + { + "tools": { + "lookup": { + "kind": "postgres-sql", + "description": "Look one thing up.", + "parameters": [{"name": "key", "type": "string"}], + "statement": "SELECT $1", + } + } + }, + "unused", + tmp_path / "trace.jsonl", + ) + headers = {"content-type": "application/json"} + if accept is not None: + headers["accept"] = accept.decode() + + # The client runs the server's lifespan, as uvicorn does. + with TestClient(application, base_url="http://127.0.0.1:5000") as client: + response = client.post( + "/mcp/catalog", + headers=headers, + json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}, + ) + + assert response.status_code == 200, response.text + assert [tool["name"] for tool in response.json()["result"]["tools"]] == ["lookup"] diff --git a/tests/harness/test_process_runtime.py b/tests/harness/test_process_runtime.py index c14f4c461..c2db0d748 100644 --- a/tests/harness/test_process_runtime.py +++ b/tests/harness/test_process_runtime.py @@ -3460,6 +3460,84 @@ def replay(): assert applied[0].source_schema_hash == discovered.fingerprint +@pytest.mark.parametrize( + ("built", "live", "applies"), + [ + (("items",), ("_alk_tool_trace", "items"), True), + (("_alk_tool_trace", "items"), ("_alk_tool_trace+", "items"), True), + (("_alk_tool_trace", "items"), ("_alk_tool_trace", "items"), True), + (("_alk_tool_trace", "items"), ("_alk_tool_trace", "items", "orders"), False), + (("_alk_tool_trace", "items"), ("_alk_tool_trace", "items+"), False), + ], + ids=[ + "log-the-build-did-not-record", + "log-widened-since-the-build", + "same-harness", + "agent-table-added", + "agent-column-added", + ], +) +def test_canonical_seed_reads_the_harness_tables_as_the_build_recorded_them( + tmp_path: Path, monkeypatch, built: tuple[str, ...], live: tuple[str, ...], applies: bool +) -> None: + """Certify-only seeds a saved World IR under whatever harness runs it: the harness's own + `_alk_` tables never change the fingerprint, any change to the agent's schema does.""" + from contextlib import nullcontext + + from fi.alk.harness.compile import postgres as compiler + from fi.alk.harness.source_model import LogicalType, SourceColumn, SourceModel, SourceTable + from fi.alk.harness.source_schema import postgres as schema + from fi.alk.harness.world_ir import WorldIR + + def table(spec: str) -> SourceTable: + name = spec.rstrip("+") + names = ("id", "extra") if spec.endswith("+") else ("id",) + columns = tuple( + SourceColumn( + name=one, + logical_type=LogicalType.STRING, + native_type="text", + nullable=one != "id", + has_default=False, + ) + for one in names + ) + return SourceTable(name=name, columns=columns, primary_key=("id",)) + + digest = "sha256:" + "b" * 64 + recorded = SourceModel.create(source_digest=digest, engine="postgres", tables=tuple(map(table, built))) + inspected = SourceModel.create(source_digest=digest, engine="postgres", tables=tuple(map(table, live))) + seed = tmp_path / "seed" + seed.mkdir() + (seed / "source-model.json").write_text(recorded.model_dump_json()) + (seed / "world-ir.json").write_text( + WorldIR.create(source_model_fingerprint=recorded.fingerprint, tables=()).model_dump_json() + ) + applied = [] + monkeypatch.setitem( + sys.modules, "psycopg", types.SimpleNamespace(connect=lambda **_: nullcontext(object())) + ) + monkeypatch.setattr(schema, "inspect_postgres", lambda *_args, **_kwargs: inspected) + monkeypatch.setattr(compiler, "apply_postgres", lambda _connection, compiled: applied.append(compiled)) + + def seed_world(): + pr.apply_postgres_world_ir( + seed / "world-ir.json", + port=14000, + dbname="baseline", + credentials=pr.EngineCredentials(username="harness", password="pw"), + source_digest=digest, + ) + + if applies: + seed_world() + assert applied[0].source_schema_hash == recorded.fingerprint + else: + with pytest.raises(pr.GenericWorldSeedError): + seed_world() + assert not applied + + def test_canonical_seed_names_a_source_model_recorded_for_other_source( tmp_path: Path, monkeypatch ) -> None: diff --git a/tests/harness/test_tool_routing_observations.py b/tests/harness/test_tool_routing_observations.py new file mode 100644 index 000000000..2546df70f --- /dev/null +++ b/tests/harness/test_tool_routing_observations.py @@ -0,0 +1,700 @@ +from __future__ import annotations + +import asyncio +import json +import time +from types import SimpleNamespace + +import pytest + +from fi.alk.harness import call_runner, chat_call_runner, chat_readiness +from fi.alk.harness.chat_readiness import TOOL_ROUTING, record_tool_routing +from fi.alk.harness.contract import AgentContract +from fi.alk.harness.world.runtime import Call + + +def _call(name, *, ok=True, refused=False, at=10.0): + return Call(name=name, arguments={}, ok=ok, refused=refused, at=at) + + +def _contract(): + return AgentContract.model_validate( + { + "agent": "shop", + "tools": [ + { + "name": "lookup_order", + "args": ["order_id"], + "arg_values": {"order_id": ["A-1"]}, + }, + {"name": "track_parcel", "args": ["parcel_id"]}, + { + "name": "issue_refund", + "args": ["order_id"], + "effects": "creates a refund row", + }, + {"name": "send_sms", "args": ["to"]}, + {"name": "get_hours"}, + ], + "tool_entrypoints": [ + { + "tool": "lookup_order", + "mode": "service", + "service": "tools-api", + "endpoint": "/orders/lookup", + }, + {"tool": "track_parcel", "mode": "service", "service": "tools-api"}, + {"tool": "issue_refund", "mode": "service", "service": "tools-api"}, + {"tool": "get_hours", "mode": "import", "module": "m", "callable": "f"}, + ], + "dependencies": [ + { + "kind": "service", + "name": "sms-gateway", + "what": "a hosted SMS API", + "used_by": ["send_sms"], + } + ], + } + ) + + +PROXIED = SimpleNamespace(endpoints={"tool_proxy": object()}) +NO_PROXY = SimpleNamespace(endpoints={}) + + +def _body(tmp_path): + return json.loads((tmp_path / "generic-harness" / TOOL_ROUTING).read_text()) + + +def test_a_tool_the_world_received_is_routed_once_with_its_outcomes(tmp_path): + record_tool_routing( + tmp_path, + "refund-1", + _contract(), + PROXIED, + # The proxy logs a call by the last segment of its path. + received=[_call("lookup"), _call("issue_refund", ok=False), _call("lookup")], + reported=[_call("lookup_order"), _call("issue_refund")], + ) + + body = _body(tmp_path) + + assert body["scenario_key"] == "refund-1" + assert body["tools"] == [ + {"tool": "issue_refund", "routed": True, "calls": 1, "outcomes": ["error"]}, + {"tool": "lookup_order", "routed": True, "calls": 2, "outcomes": ["ok"]}, + ] + assert body["unobserved"] == [] + + +def test_a_tool_the_agent_reports_but_the_world_never_received_is_not_routed(tmp_path): + record_tool_routing( + tmp_path, + "refund-1", + _contract(), + PROXIED, + received=[_call("lookup")], + reported=[_call("lookup_order"), _call("track_parcel")], + ) + + assert _body(tmp_path)["tools"] == [ + {"tool": "lookup_order", "routed": True, "calls": 1, "outcomes": ["ok"]}, + {"tool": "track_parcel", "routed": False, "reason": "not_received"}, + ] + + +def test_a_tool_the_agent_runs_itself_is_never_counted_routed(tmp_path): + # In-process tools, and service tools with no proxy in front: no world record exists. + record_tool_routing( + tmp_path, + "refund-1", + _contract(), + NO_PROXY, + received=[], + reported=[_call("get_hours"), _call("track_parcel")], + ) + + body = _body(tmp_path) + + assert body["tools"] == [] + assert body["unobserved"] == [ + {"tool": "get_hours", "reason": "no_world_record"}, + {"tool": "track_parcel", "reason": "no_world_record"}, + ] + + +def test_a_tool_the_mcp_toolbox_serves_is_logged_by_the_world(tmp_path): + served = tmp_path / "bundle" / "generated" / "mcp-toolbox" + served.mkdir(parents=True) + (served / "toolbox.json").write_text(json.dumps({"tools": {"get_hours": {}, "track_parcel": {}}})) + runtime = SimpleNamespace(endpoints={"mcp_toolbox": object()}) + + record_tool_routing( + tmp_path, + "s-1", + _contract(), + runtime, + received=[_call("get_hours")], + reported=[_call("get_hours"), _call("track_parcel")], + bundle=tmp_path / "bundle", + ) + + assert _body(tmp_path)["tools"] == [ + {"tool": "get_hours", "routed": True, "calls": 1, "outcomes": ["ok"]}, + # Reported, and the toolbox would have logged it: the call went elsewhere. + {"tool": "track_parcel", "routed": False, "reason": "not_received"}, + ] + + +class _World: + def __init__(self): + self.calls = [] + + def handle_tool_call(self, call): + self.calls.append(_call(call["name"])) + return SimpleNamespace(content="{}") + + +def _requests( + tmp_path, + monkeypatch, + *, + runtime=PROXIED, + bounds=None, + candidate="sha256:one", + **behaviour, +): + """Each request is answered as ``behaviour`` says, per tool: ``world`` (the call goes + through the proxy), ``reported`` (the agent says it ran it; nothing reached the world), + ``returned`` (the agent returns the call for the harness to run), ``fail``.""" + world_log: list = [] + file_trace: list = [] + sent: list = [] + + class Wrapper: + def __init__(self): + self.returned = False + + async def call(self, agent_input): + text = agent_input.new_message["content"] + sent.append(text) + tool = next( + (one for one in behaviour.get("names", ()) if f"your {one} tool" in text), + None, + ) + how = behaviour.get(tool) if tool else None + if how == "fail": + raise RuntimeError("agent down") + if how == "world": + world_log.append(_call(tool, at=1e12)) + file_trace.append(_call(tool)) + if how == "reported": + file_trace.append(_call(tool)) + calls = [] + if how == "returned" and not self.returned: + self.returned = True + calls = [{"id": "c1", "name": tool, "arguments": {}}] + return SimpleNamespace( + content="done", tool_calls=calls, tool_responses=[], metadata={} + ) + + monkeypatch.setattr( + chat_readiness, "create_chat_wrapper", lambda *a, **k: Wrapper() + ) + monkeypatch.setattr(chat_call_runner, "_tool_world", lambda *a, **k: _World()) + monkeypatch.setattr( + call_runner, "_collect_file_tool_calls", lambda runtime: tuple(file_trace) + ) + monkeypatch.setattr( + call_runner, "_collect_tool_trace_calls", lambda runtime: tuple(world_log) + ) + record_tool_routing( + tmp_path, + "s-1", + _contract(), + runtime, + received=[_call("lookup")], + reported=[_call("lookup_order")], + ) + asyncio.run( + chat_readiness.request_uncovered_tools( + interface=None, + endpoint="http://agent", + runtime=runtime, + contract=_contract(), + authoring=tmp_path, + bundle=tmp_path, + source=None, + secret_values={}, + candidate_hash=candidate, + **(bounds or {}), + ) + ) + return _body(tmp_path), sent + + +NAMES = ("track_parcel", "issue_refund", "get_hours") + + +def test_each_uncalled_tool_is_asked_once_and_routed_only_when_the_world_receives_it( + tmp_path, monkeypatch +): + body, sent = _requests( + tmp_path, + monkeypatch, + names=NAMES, + track_parcel="world", + issue_refund="reported", + get_hours="returned", + ) + + assert body["tools"] == [ + # The agent returned the call and the harness ran it at the world. + {"tool": "get_hours", "routed": True, "calls": 1, "outcomes": ["ok"], "source": "tool_request"}, + {"tool": "lookup_order", "routed": True, "calls": 1, "outcomes": ["ok"]}, + {"tool": "track_parcel", "routed": True, "calls": 1, "outcomes": ["ok"], "source": "tool_request"}, + ] + assert body["unobserved"] == [ + {"tool": "issue_refund", "reason": "has_effects"}, + {"tool": "send_sms", "reason": "reaches_outside"}, + ] + assert len(sent) == 3 # two tools, one continuation after the returned call + assert not any( + "send_sms" in text or "lookup_order" in text or "issue_refund" in text for text in sent + ) + + +def test_a_tool_the_agent_says_it_ran_that_the_world_never_received_is_not_routed( + tmp_path, monkeypatch +): + body, _ = _requests(tmp_path, monkeypatch, names=NAMES, track_parcel="reported") + + assert { + "tool": "track_parcel", "routed": False, "reason": "not_received", "source": "tool_request" + } in body["tools"] + + +def test_a_request_the_agent_cannot_answer_or_ignores_leaves_the_tool_unobserved( + tmp_path, monkeypatch +): + body, _ = _requests(tmp_path, monkeypatch, names=NAMES, track_parcel="fail") + + assert {"tool": "track_parcel", "reason": "request_failed"} in body["unobserved"] + assert {"tool": "get_hours", "reason": "not_called"} in body["unobserved"] + + +def test_requests_past_the_limit_are_recorded_as_not_asked(tmp_path, monkeypatch): + body, sent = _requests( + tmp_path, monkeypatch, bounds={"limit": 1}, names=NAMES, track_parcel="world" + ) + + assert len(sent) == 1 + asked = [one["tool"] for one in body["tools"] if one.get("source") == "tool_request"] + left = sorted(one["tool"] for one in body["unobserved"] if one["reason"] == "not_asked") + assert len(asked) + len(left) == 2 # issue_refund has effects and is never asked + assert len(left) == 1 + assert all(one.get("reason") != "not_asked" for one in body["tools"]) + + +def test_no_request_is_sent_once_the_time_is_spent(tmp_path, monkeypatch): + body, sent = _requests( + tmp_path, monkeypatch, bounds={"budget": 0.0}, names=NAMES, track_parcel="world" + ) + + assert sent == [] + assert sorted(one["tool"] for one in body["unobserved"] if one["reason"] == "not_asked") == [ + "get_hours", + "track_parcel", + ] + + +def test_the_request_names_the_tool_and_an_example_from_its_contract(): + text = chat_readiness._tool_request(_contract().tools[0]) + + assert ( + "lookup_order" in text + and "order_id='A-1'" in text + and "disposable test world" in text + ) + + +def test_an_argument_without_an_example_takes_a_plain_value_of_its_type(): + tool = AgentContract.model_validate( + { + "agent": "a", + "tools": [ + { + "name": "apply", + "args": ["item_id", "note"], + "arg_types": {"item_id": "integer"}, + } + ], + } + ).tools[0] + + assert "item_id=1, note='test'" in chat_readiness._tool_request(tool) + + +def test_a_later_attempt_of_the_check_reuses_the_answers_and_stays_within_the_limit( + tmp_path, monkeypatch +): + first, sent_first = _requests( + tmp_path, monkeypatch, bounds={"limit": 2}, names=NAMES, track_parcel="world" + ) + # A retry writes the qualification's routing again, then asks for the rest. + again, sent_again = _requests( + tmp_path, monkeypatch, bounds={"limit": 2}, names=NAMES, track_parcel="world" + ) + + assert len(sent_first) == 2 + assert sent_again == [] + assert again["tools"] == first["tools"] + assert again["unobserved"] == first["unobserved"] + + +@pytest.mark.parametrize("runtime", [PROXIED, NO_PROXY], ids=["world_logs_it", "no_world_record"]) +def test_a_tool_with_effects_is_never_asked_even_when_the_world_logs_its_calls( + tmp_path, monkeypatch, runtime +): + body, sent = _requests( + tmp_path, monkeypatch, runtime=runtime, names=NAMES, issue_refund="world" + ) + + assert {"tool": "issue_refund", "reason": "has_effects"} in body["unobserved"] + assert not any("issue_refund" in text for text in sent) + assert all(one["tool"] != "issue_refund" for one in body["tools"]) + + +def test_one_request_stays_within_the_budget_however_many_times_the_agent_is_called( + tmp_path, monkeypatch +): + """Each wrapper call is slow and returns another tool call, so ``say()`` would call the + agent eight times; the whole request must still end within the shared budget.""" + calls = [] + + class Slow: + async def call(self, agent_input): + calls.append(1) + await asyncio.sleep(0.2) + return SimpleNamespace( + content="", + tool_calls=[{"id": f"c{len(calls)}", "name": "get_hours", "arguments": {}}], + tool_responses=[], + metadata={}, + ) + + monkeypatch.setattr(chat_readiness, "create_chat_wrapper", lambda *a, **k: Slow()) + monkeypatch.setattr(chat_call_runner, "_tool_world", lambda *a, **k: _World()) + monkeypatch.setattr(call_runner, "_collect_file_tool_calls", lambda runtime: ()) + monkeypatch.setattr(call_runner, "_collect_tool_trace_calls", lambda runtime: ()) + record_tool_routing(tmp_path, "s-1", _contract(), PROXIED, received=[], reported=[]) + started = time.monotonic() + asyncio.run( + chat_readiness.request_uncovered_tools( + interface=None, + endpoint="http://agent", + runtime=PROXIED, + contract=_contract(), + authoring=tmp_path, + bundle=tmp_path, + source=None, + secret_values={}, + candidate_hash="sha256:one", + budget=0.5, + timeout=0.5, + ) + ) + + assert time.monotonic() - started < 1.0 + assert len(calls) < 8 + body = _body(tmp_path) + assert {"tool": "lookup_order", "reason": "request_failed"} in body["unobserved"] + + +def test_a_request_runs_only_the_asked_tool_at_the_world(tmp_path, monkeypatch): + """Asked for one read-only tool, the agent returns a tool with effects instead: the world + refuses it rather than running it, and the routing file names the refusal.""" + world = _World() + + class Agent: + def __init__(self): + self.turn = 0 + + async def call(self, agent_input): + self.turn += 1 + if self.turn == 1: + return SimpleNamespace( + content="", + tool_calls=[ + {"id": "c1", "name": "issue_refund", "arguments": {"order_id": "A-1"}} + ], + tool_responses=[], + metadata={}, + ) + return SimpleNamespace(content="done", tool_calls=[], tool_responses=[], metadata={}) + + monkeypatch.setattr(chat_readiness, "create_chat_wrapper", lambda *a, **k: Agent()) + monkeypatch.setattr(chat_call_runner, "_tool_world", lambda *a, **k: world) + monkeypatch.setattr(call_runner, "_collect_file_tool_calls", lambda runtime: ()) + monkeypatch.setattr(call_runner, "_collect_tool_trace_calls", lambda runtime: ()) + record_tool_routing(tmp_path, "s-1", _contract(), PROXIED, received=[], reported=[]) + + asyncio.run( + chat_readiness.request_uncovered_tools( + interface=None, + endpoint="http://agent", + runtime=PROXIED, + contract=_contract(), + authoring=tmp_path, + bundle=tmp_path, + source=None, + secret_values={}, + candidate_hash="sha256:one", + limit=1, + ) + ) + + ran = [call.name for call in world.calls if not getattr(call, "refused", False)] + assert "issue_refund" not in ran + body = json.loads((tmp_path / "generic-harness" / TOOL_ROUTING).read_text()) + assert body["refused_during_requests"][0]["refused"] == ["issue_refund"] + + +def _shared_segment_contract(): + return AgentContract.model_validate( + { + "agent": "directory", + "tools": [ + {"name": "find_users", "args": ["q"]}, + {"name": "find_orders", "args": ["q"]}, + ], + "tool_entrypoints": [ + { + "tool": "find_users", + "mode": "service", + "service": "api", + "endpoint": "/users/search", + }, + { + "tool": "find_orders", + "mode": "service", + "service": "api", + "endpoint": "/orders/search", + }, + ], + } + ) + + +def test_a_proxied_call_is_credited_by_its_full_path_when_two_paths_share_a_last_segment( + tmp_path, +): + record_tool_routing( + tmp_path, + "s-1", + _shared_segment_contract(), + PROXIED, + received=[ + Call(name="search", arguments={}, at=10.0, endpoint="/orders/search?page=1") + ], + reported=[_call("find_orders")], + ) + + body = _body(tmp_path) + assert body["tools"] == [ + {"tool": "find_orders", "routed": True, "calls": 1, "outcomes": ["ok"]} + ] + assert body["unobserved"] == [] + + +def test_a_record_naming_only_a_shared_last_segment_credits_no_tool(tmp_path): + # An older proxy logged the last path segment alone; either tool could have received it. + record_tool_routing( + tmp_path, + "s-1", + _shared_segment_contract(), + PROXIED, + received=[_call("search")], + reported=[_call("find_orders")], + ) + + body = _body(tmp_path) + assert body["tools"] == [] + assert body["unobserved"] == [ + {"tool": "find_orders", "reason": "ambiguous_world_record"} + ] + + +def test_the_tool_proxy_logs_the_full_path_of_each_call(monkeypatch): + import importlib.util + import sys + import threading + import urllib.request + from contextlib import contextmanager + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + from pathlib import Path + + class Upstream(BaseHTTPRequestHandler): + def log_message(self, *_args): + return + + def do_POST(self): # noqa: N802 + self.rfile.read(int(self.headers.get("content-length") or 0)) + self.send_response(200) + self.send_header("content-length", "2") + self.end_headers() + self.wfile.write(b"{}") + + logged = [] + + @contextmanager + def connect(*_args, **_kwargs): + yield SimpleNamespace( + execute=lambda statement, values: logged.append((statement, values)) + ) + + upstream = ThreadingHTTPServer(("127.0.0.1", 0), Upstream) + threading.Thread(target=upstream.serve_forever, daemon=True).start() + monkeypatch.setitem(sys.modules, "psycopg", SimpleNamespace(connect=connect)) + monkeypatch.setenv("PORT", "0") + monkeypatch.setenv("UPSTREAM_URL", f"http://127.0.0.1:{upstream.server_port}") + monkeypatch.setenv("DATABASE_URL", "unused") + path = Path(chat_readiness.__file__).with_name("tool_trace_proxy.py") + spec = importlib.util.spec_from_file_location("tool_trace_proxy_under_test", path) + proxy_module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(proxy_module) + proxy = ThreadingHTTPServer(("127.0.0.1", 0), proxy_module.Handler) + threading.Thread(target=proxy.serve_forever, daemon=True).start() + try: + urllib.request.urlopen( + urllib.request.Request( + f"http://127.0.0.1:{proxy.server_port}/orders/search?page=1", + data=b"{}", + method="POST", + ), + timeout=5, + ).read() + finally: + proxy.shutdown() + upstream.shutdown() + + statement, values = logged[0] + assert "endpoint" in statement + assert values[0] == "search" + assert "/orders/search" in values + + +def test_a_blocking_tool_handler_cannot_outlast_the_request_deadline( + tmp_path, monkeypatch +): + """The world runs a returned call synchronously; a handler that blocks past the deadline + must end the request as unanswered, not let the handler finish and count as routed.""" + + class Blocking(_World): + def handle_tool_call(self, call, **context): + time.sleep(1.0) + return super().handle_tool_call(call) + + class Agent: + def __init__(self): + self.turn = 0 + + async def call(self, agent_input): + self.turn += 1 + calls = ( + [{"id": "c1", "name": "track_parcel", "arguments": {}}] + if self.turn == 1 + else [] + ) + return SimpleNamespace( + content="done", tool_calls=calls, tool_responses=[], metadata={} + ) + + monkeypatch.setattr(chat_readiness, "create_chat_wrapper", lambda *a, **k: Agent()) + monkeypatch.setattr(chat_call_runner, "_tool_world", lambda *a, **k: Blocking()) + monkeypatch.setattr(call_runner, "_collect_file_tool_calls", lambda runtime: ()) + monkeypatch.setattr(call_runner, "_collect_tool_trace_calls", lambda runtime: ()) + record_tool_routing( + tmp_path, + "s-1", + _contract(), + PROXIED, + received=[_call("lookup")], + reported=[_call("lookup_order")], + ) + started = time.monotonic() + asyncio.run( + chat_readiness.request_uncovered_tools( + interface=None, + endpoint="http://agent", + runtime=PROXIED, + contract=_contract(), + authoring=tmp_path, + bundle=tmp_path, + source=None, + secret_values={}, + candidate_hash="sha256:one", + timeout=0.1, + limit=1, + ) + ) + + assert time.monotonic() - started < 0.8 + body = _body(tmp_path) + assert all(one["tool"] != "track_parcel" for one in body["tools"]) + assert {"tool": "track_parcel", "reason": "request_failed"} in body["unobserved"] + + +def test_a_repaired_candidate_never_reuses_the_routing_proof_of_the_one_before( + tmp_path, monkeypatch +): + first, _ = _requests(tmp_path, monkeypatch, names=NAMES, track_parcel="world") + # The repair changed the candidate; the agent now never reaches the world for the tool. + again, sent = _requests( + tmp_path, + monkeypatch, + candidate="sha256:repaired", + names=NAMES, + track_parcel="reported", + ) + + assert { + "tool": "track_parcel", + "routed": True, + "calls": 1, + "outcomes": ["ok"], + "source": "tool_request", + } in (first["tools"]) + assert any("track_parcel" in text for text in sent) + assert { + "tool": "track_parcel", + "routed": False, + "reason": "not_received", + "source": "tool_request", + } in again["tools"] + + +def test_a_repaired_candidate_past_the_shared_budget_leaves_the_tool_unobserved( + tmp_path, monkeypatch +): + _requests( + tmp_path, monkeypatch, bounds={"limit": 2}, names=NAMES, track_parcel="world" + ) + again, sent = _requests( + tmp_path, + monkeypatch, + candidate="sha256:repaired", + bounds={"limit": 2}, + names=NAMES, + track_parcel="world", + ) + + assert sent == [] + assert all(one.get("source") != "tool_request" for one in again["tools"]) + assert sorted( + one["tool"] for one in again["unobserved"] if one["reason"] == "not_asked" + ) == [ + "get_hours", + "track_parcel", + ]