diff --git a/.github/workflows/sdk-smoke.yml b/.github/workflows/sdk-smoke.yml index 5cecf3a48..8589e0e0d 100644 --- a/.github/workflows/sdk-smoke.yml +++ b/.github/workflows/sdk-smoke.yml @@ -36,7 +36,7 @@ jobs: - name: Install test dependencies # Exercise the minimum LiveKit version supported by pyproject.toml. run: >- - python -m pip install '.[livekit]' pytest pytest-asyncio + python -m pip install '.[livekit]' pytest pytest-asyncio 'hypothesis>=6.140' 'livekit-agents==1.8.3' 'livekit-plugins-cartesia==1.8.3' 'livekit-plugins-deepgram==1.8.3' diff --git a/src/fi/alk/harness/agent_reading.py b/src/fi/alk/harness/agent_reading.py index 7ce74e0b5..e8aac99b2 100644 --- a/src/fi/alk/harness/agent_reading.py +++ b/src/fi/alk/harness/agent_reading.py @@ -22,7 +22,7 @@ from .backends import StageDone from .cli import guidance_text from .failure_file import FAILURE_FILE -from .job import FailureDomain, HarnessFailure, HarnessJob, ProviderExecutionMode, SourceKind +from .job import FailureDomain, HarnessFailure, HarnessJob, SourceKind, provider_profile_source from .reading_entry import ( EXIT_RETRYABLE, READING_JOB_INVALID, @@ -75,13 +75,14 @@ def _soft_limit() -> float: def _check_job(job: HarnessJob) -> HarnessFailure | None: - if job.agent.mode in (ProviderExecutionMode.PROVIDER_IMPORT, ProviderExecutionMode.CONNECT_ONLY): - # The build reads these agents from the provider's definition, which a reading never loads. + if provider_profile_source(job.agent) is not None: + # The build adds a profile to what it reads for these agents, which a reading never loads. return reading_failure( FailureDomain.PLATFORM_SYNC, READING_JOB_INVALID, - "A reading without a build reads an agent from its repository only. A " - f"{job.agent.mode.value} job's agent is defined by its provider, so it needs a build.", + "A reading without a build reads an agent from its repository only. This job's agent " + "is also described outside its repository (by its provider, or by a prompt in the " + "job), so it needs a build.", retryable=False, ) if job.source.kind in (SourceKind.GITHUB, SourceKind.ARCHIVE): @@ -201,7 +202,8 @@ async def _stop( await asyncio.wait({task}, timeout=wait) # Closed before the files are read, so a late tool call is unlikely to save after the # decision. The close is bounded, though: a session that hangs closing must not keep the - # reading from deciding, and one that outlives the bound could still act afterwards. + # reading from deciding; a session that outlives the bound is cancelled, with every other + # task, before the reading returns (`_stop_the_rest`). task.cancel() closing = min(wait, _CLOSE_SECONDS) await asyncio.wait({task}, timeout=closing) @@ -211,6 +213,23 @@ async def _stop( logger.warning("the stopped reading ended with an error", exc_info=task.exception()) +def _stop_the_rest() -> None: + # Every task on this loop but this one is cancelled, ahead of the next loop turn, so no + # late tool call can change the files the reading decided on. asyncio.run's shutdown + # reaches the same tasks a moment later and cancels any still pending a second time, so a + # cleanup that catches the first cancel and then awaits may be cut short. It relies on `_read` running as asyncio.run's main task. + # The precondition: every tool call runs as a task on this loop and honours a cancel. + # The reading's Gemini tools are pinned to run on its loop (ADK's tool_thread_pool_config + # applies only to live mode in 2.7.1, which the reading doesn't use); its Claude tools run + # in the SDK's MCP server task on the same loop. The contract save is pinned to write + # nothing when cancelled at any of its waits. A tool that runs on another thread's loop, + # or swallows its cancel, is not covered by the guard. Shown against the locked versions. + this = asyncio.current_task() + for other in asyncio.all_tasks(): + if other is not this: + other.cancel() + + def _timed_out(output: Path, soft_limit: float) -> int: write_reading_failure( reading_failure( @@ -338,6 +357,7 @@ async def _read(job: HarnessJob, source: Path, output: Path, *, interrupt_wait: ) code, complete = document(job, Path(source), output), False finally: + _stop_the_rest() # The work owns the final cost file, on every way out. write_reading_cost(output, complete=complete) return code diff --git a/src/fi/alk/harness/amend.py b/src/fi/alk/harness/amend.py index 841568394..42e1e4302 100644 --- a/src/fi/alk/harness/amend.py +++ b/src/fi/alk/harness/amend.py @@ -19,12 +19,18 @@ from __future__ import annotations -import json from pathlib import Path -from .contract import AgentContract, known_modalities, validate_contract +from ..providers.agent import _redact +from .contract import ( + CONTRACT_FILE, + AgentContract, + known_modalities, + validate_contract, + write_contract, +) -CONTRACT = "contract.json" +CONTRACT = CONTRACT_FILE def widen( @@ -78,12 +84,7 @@ def widen( contract.amendments.pop() return False, "the amended contract would not be valid: " + "; ".join(problems) - destination = Path(destination) - destination.mkdir(parents=True, exist_ok=True) - (destination / CONTRACT).write_text( - json.dumps(contract.model_dump(), indent=2, ensure_ascii=False), - encoding="utf-8", - ) + _persist(contract, destination) return True, ( f"{tool_name}.{argument} now accepts {', '.join(fresh)}. " f"{len(contract.amendments)} amendment(s) recorded on the contract." @@ -104,8 +105,12 @@ def add_rule( return False, "no rule given" if not why.strip(): return False, "say why: an unexplained rule is indistinguishable from a guess" - if any(rule.lower() == existing.lower() for existing in contract.hard_constraints): - return False, f"the agent already has that rule: {rule}" + # Both sides are compared as they are saved: a contract held from an older save can still + # hold its rules raw. One call, so a refused URL is logged once. The reply names the rule as + # saved too, since the chat sends it on to the platform. + cleaned, *existing = _redact([rule, *contract.hard_constraints]) + if cleaned.lower() in (one.lower() for one in existing): + return False, f"the agent already has that rule: {cleaned}" contract.hard_constraints.append(rule) contract.amendments.append(f"rule added — {rule}: {why.strip()}") @@ -115,12 +120,7 @@ def add_rule( contract.amendments.pop() return False, "the amended contract would not be valid: " + "; ".join(problems) - destination = Path(destination) - destination.mkdir(parents=True, exist_ok=True) - (destination / CONTRACT).write_text( - json.dumps(contract.model_dump(), indent=2, ensure_ascii=False), - encoding="utf-8", - ) + _persist(contract, destination) return True, ( f"added. The agent now has {len(contract.hard_constraints)} rules, and this one is " "graded from here on." @@ -166,12 +166,7 @@ def set_modality( contract.amendments.pop() return False, "the amended contract would not be valid: " + "; ".join(problems) - destination = Path(destination) - destination.mkdir(parents=True, exist_ok=True) - (destination / CONTRACT).write_text( - json.dumps(contract.model_dump(), indent=2, ensure_ascii=False), - encoding="utf-8", - ) + _persist(contract, destination) reached = { "voice": "a call placed to the agent where it is hosted, its own tools answered over a " "webhook", @@ -191,25 +186,23 @@ def drop_rule( """ if not why.strip(): return False, "say why: removing a rule changes what is being graded" + # Both sides are compared as they are saved: a contract held from an older save can still + # hold its rules raw. One call, so a refused URL is logged once. + wanted, *saved_rules = _redact([rule.strip(), *contract.hard_constraints]) + wanted = wanted.lower() + held = list(zip(contract.hard_constraints, (one.lower() for one in saved_rules))) match = next( - ( - existing - for existing in contract.hard_constraints - if existing.lower() == rule.strip().lower() - ), + (existing for existing, as_saved in held if as_saved == wanted), None, ) or next( - ( - existing - for existing in contract.hard_constraints - if rule.strip().lower() in existing.lower() - ), + (existing for existing, as_saved in held if wanted in as_saved), None, ) if match is None: + # The rules as saved, since the chat sends this reply on to the platform. return False, ( "no rule like that. It has:\n - " - + "\n - ".join(contract.hard_constraints) + + "\n - ".join(saved_rules) ) contract.hard_constraints.remove(match) contract.amendments.append(f"rule removed — {match}: {why.strip()}") @@ -322,12 +315,9 @@ def drop_tool_references(contract: AgentContract) -> list[str]: def _persist(contract: AgentContract, destination: Path) -> None: - destination = Path(destination) - destination.mkdir(parents=True, exist_ok=True) - (destination / CONTRACT).write_text( - json.dumps(contract.model_dump(), indent=2, ensure_ascii=False), - encoding="utf-8", - ) + """Save the contract the stage is holding. Its free text is cleaned in place as it is saved, + so the stage goes on working from what was saved.""" + write_contract(contract, Path(destination)) def not_offered(contract: AgentContract, candidates: dict[str, set[str]]) -> list[str]: diff --git a/src/fi/alk/harness/authoring_entrypoint.py b/src/fi/alk/harness/authoring_entrypoint.py index 40012628f..f861c578f 100644 --- a/src/fi/alk/harness/authoring_entrypoint.py +++ b/src/fi/alk/harness/authoring_entrypoint.py @@ -25,15 +25,19 @@ from .agent_type_check import EXIT_REFUSED, failure_message from .cli import _auto from .failure_file import FAILURE_FILE -from .job import HarnessJob, ProviderExecutionMode, SourceKind +from .job import ( + PROVIDER_MANAGED_CONNECTORS as PROVIDER_MANAGED_CONNECTORS, + HarnessJob, + ProviderExecutionMode, + SourceKind, + provider_profile_source, +) from .hosted_compose_bootstrap import initialize_hosted_compose from .provider_import import inspect_provider_target from .scenarios import load as load_written from .source_folder import refuse_unusable_folder, saved_runtime from .understand import PROVIDER_IMPORT_PROFILE_PATH_ENV -PROVIDER_MANAGED_CONNECTORS = frozenset({"vapi", "retell", "retell_chat"}) - def _persist_authored_scenario_count( job_path: Path, job: HarnessJob, output: Path @@ -468,10 +472,8 @@ def _load_provider_import_profile( secrets_path: Path | None, profile_cache_path: Path | None = None, ) -> dict[str, object] | None: - if ( - job.agent.mode is ProviderExecutionMode.CONNECT_ONLY - and job.agent.connector.strip().lower() == "phone" - ): + source = provider_profile_source(job.agent) + if source == "phone": # A dial-in number has no provider API to inspect. The supplied prompt is the only # behavioral source of truth; never pretend we discovered tools or implementation code. return { @@ -481,10 +483,7 @@ def _load_provider_import_profile( "system_prompt": str(job.agent.config["target_system_prompt"]), "tools": [], } - if ( - job.agent.mode is ProviderExecutionMode.CONNECT_ONLY - and job.agent.connector.strip().lower() == "livekit" - ): + if source == "livekit": if secrets_path is not None: secrets_path.unlink(missing_ok=True) return { @@ -494,14 +493,7 @@ def _load_provider_import_profile( "system_prompt": str(job.agent.config["target_system_prompt"]), "tools": [], } - inspect_connect_only_provider = ( - job.agent.mode is ProviderExecutionMode.CONNECT_ONLY - and job.agent.connector.strip().lower() in PROVIDER_MANAGED_CONNECTORS - ) - if ( - job.agent.mode is not ProviderExecutionMode.PROVIDER_IMPORT - and not inspect_connect_only_provider - ): + if source is None: return None if profile_cache_path is not None and profile_cache_path.is_file(): cached = json.loads(profile_cache_path.read_text(encoding="utf-8")) diff --git a/src/fi/alk/harness/authoring_runtime_validation.py b/src/fi/alk/harness/authoring_runtime_validation.py index 1fc70d4dc..354749683 100644 --- a/src/fi/alk/harness/authoring_runtime_validation.py +++ b/src/fi/alk/harness/authoring_runtime_validation.py @@ -1427,6 +1427,27 @@ def _invariants_the_ir_still_breaks(world, authoring: Path) -> list[str]: return [] +def _promote_data_store(contract, authoring: Path, candidate_dir: Path) -> int: + """Copy only ``data_store`` from a candidate reading onto the saved contract, and save it. + + Returns 0 when it was promoted, 1 when the candidate has none to give. + """ + from .contract import AgentContract, write_contract + + candidate_path = candidate_dir / "contract.json" + if not candidate_path.is_file(): + return 1 + candidate = AgentContract.model_validate_json( + candidate_path.read_text(encoding="utf-8") + ) + if candidate.data_store is None or not candidate.data_store.kind.strip(): + return 1 + payload = contract.model_dump() + payload["data_store"] = candidate.data_store.model_dump() + write_contract(AgentContract.model_validate(payload), authoring) + return 0 + + async def validate_and_repair( job, source: Path, @@ -1509,24 +1530,11 @@ async def repair(phase, guidance): ], ) ) - candidate_path = Path(candidate_dir) / "contract.json" - if status != 0 or not candidate_path.is_file(): - return 1 - candidate = AgentContract.model_validate_json( - candidate_path.read_text(encoding="utf-8") - ) - if ( - candidate.data_store is None - or not candidate.data_store.kind.strip() - ): + if status != 0: return 1 - payload = contract.model_dump() - payload["data_store"] = candidate.data_store.model_dump() - repaired = AgentContract.model_validate(payload) - contract_path.write_text( - repaired.model_dump_json(indent=2) + "\n", encoding="utf-8" + return _promote_data_store( + contract, authoring, Path(candidate_dir) ) - return 0 if generic and _needs_runtime_plan_repair(_active_repair_diagnostics): from .contract import AgentContract from .runtime_repair import request_runtime_plan_patch diff --git a/src/fi/alk/harness/cli.py b/src/fi/alk/harness/cli.py index 60971fc5f..9c3ba928c 100644 --- a/src/fi/alk/harness/cli.py +++ b/src/fi/alk/harness/cli.py @@ -19,6 +19,7 @@ from pathlib import Path from typing import Any +from ..providers.agent import _redact from . import spend from . import observability from .build import open_stage as build_stage @@ -62,7 +63,9 @@ def _source_root(destination: Path, explicit: str = "") -> str: def _render(event: Event) -> None: - line = event.line() + # The platform keeps the tail of this output, so a URL's password, query and fragment are + # dropped first, in one call per event, as emit_model_event drops them from the job's events. + line = _redact(event.line()) if event.kind == TEXT: print(line, end="", flush=True) else: @@ -109,7 +112,11 @@ def receive(event: Event) -> None: emitted.append(event.text) await stage.say(content, on_event=receive) - text = "".join(emitted).strip() or "I’m still working on the active authoring stage." + # Sent to the platform's chat as the hosted builder's replies are, so cleaned the same way. + text = ( + _redact("".join(emitted).strip()) + or "I’m still working on the active authoring stage." + ) await client.emit( "assistant_message", message_id=str(uuid.uuid4()), @@ -1015,12 +1022,15 @@ def emit_model_event(event: Event) -> None: for key in ("target", "label", "call_id", "is_error", "path", "outcome", "cost_usd") if key in detail } + # The job response carries these, so a URL's password or query is dropped first, in + # one call, before the text is cut. + text, safe_detail = _redact([event.text or None, safe_detail]) emit( "harness.activity", str(detail.get("stage") or "authoring"), event_kind=event.kind, tool=event.tool or None, - text=event.text[:2000] if event.text else None, + text=text[:2000] if text else None, detail=safe_detail, ) diff --git a/src/fi/alk/harness/contract.py b/src/fi/alk/harness/contract.py index 3fc5423bf..b806786e3 100644 --- a/src/fi/alk/harness/contract.py +++ b/src/fi/alk/harness/contract.py @@ -11,9 +11,13 @@ from __future__ import annotations +import collections.abc import json +import os import re import shlex +import threading +import typing from pathlib import Path from typing import Any, Literal @@ -25,6 +29,7 @@ CompletedToolRows, ) +from ..providers.agent import _redact from .observations import HttpObservation # How a person reaches an agent. This decides how it is later run — voice goes out as a live @@ -1392,6 +1397,222 @@ def adoptable(self, tool: str) -> bool: return bool(found and found.mode in ("import", "construct", "service")) +# Which parts of the contract are prose, written by the reader in its own words, and which are +# read as written by the build, the world or the runtime. Prose is cleaned of URL credentials +# (userinfo, query and fragment) whenever the contract is saved; the rest is never changed. +# A test walks every model the contract reaches and fails on a field in neither tuple. +FREE_TEXT = ( + "AgentContract.one_liner", + "AgentContract.system_prompt_excerpt", + "AgentContract.hard_constraints", + "AgentContract.implementation", + "AgentContract.real_use_cases", + "AgentContract.caller_identity", + "AgentContract.notes", + "AgentContract.open_questions", + "AgentContract.amendments", + "ToolSpec.description", + "ToolSpec.refuses", + "ToolSpec.effects", + "ToolEntry.notes", + "Dependency.what", +) +# A field that holds models is kept for itself, and its models' fields are classified in their +# own right. A name without a dot is a whole technical model, every field of it kept. +KEPT_AS_WRITTEN = ( + "AgentContract.agent", + "AgentContract.modality", + "AgentContract.call_direction", + "AgentContract.agent_languages", + "AgentContract.conversational", + "AgentContract.system_prompt_from", + "AgentContract.unconfirmed_prompt_from", + # The model name the code cites, with its file and line: read as written, like the prompt lines. + "AgentContract.model", + "AgentContract.tools", + "AgentContract.data_schema", + "AgentContract.base_environment", + "AgentContract.dependencies", + "AgentContract.runtime_dependencies", + "AgentContract.tool_entrypoints", + # world/runtime.py matches its quoted markers against tool results, so cleaning it would + # change what counts as a refusal. + "AgentContract.refusal_signature", + "AgentContract.data_store", + "AgentContract.runtime", + "AgentContract.chosen_evals", + "AgentContract.instrumentation", + "ToolSpec.name", + "ToolSpec.args", + "ToolSpec.arg_types", + # The world answers with these as written. + "ToolSpec.arg_values", + "ToolSpec.requires", + "ToolSpec.inputs_from", + "ToolEntry.tool", + "ToolEntry.mode", + "ToolEntry.module", + "ToolEntry.callable", + "ToolEntry.factory", + "ToolEntry.first_arg", + "ToolEntry.service", + # The tool proxy calls it as written. + "ToolEntry.endpoint", + "ToolEntry.method", + "Dependency.name", + "Dependency.kind", + "Dependency.used_by", + "Dependency.engine", + "Dependency.version", + "Dependency.reached", + "PromptSource", + "ModelSource", + "Reached", + "DataStore", + "Runtime", + "HttpObservation", + "ObservationParameter", + "RuntimeSecretFile", + "RuntimeConfigFile", + "RuntimeInterface", + "CompletedToolEvents", + "CompletedToolFailureEvent", + "CompletedToolRows", + "LocalHttpAuth", + "RuntimeHttpFailureRule", + "RuntimeHttpSetupRequest", +) +CONTRACT_FILE = "contract.json" +_SEQUENCES = ( + list, + tuple, + set, + frozenset, + collections.abc.Sequence, + collections.abc.MutableSequence, +) +_MAPPINGS = (dict, collections.abc.Mapping, collections.abc.MutableMapping) + + +def _walks_as_sequence(origin: Any) -> bool: + """Whether _holders walks a field of this origin as a sequence: any subclass of those above + (an abc registration counts), but never text.""" + return ( + isinstance(origin, type) + and issubclass(origin, _SEQUENCES) + and not issubclass(origin, (str, bytes, bytearray)) + ) + + +def _walks_as_mapping(origin: Any) -> bool: + """Whether _holders walks a field of this origin as a mapping: any subclass of those above, + such as an OrderedDict (change review r3, B-L1).""" + return isinstance(origin, type) and issubclass(origin, _MAPPINGS) + + +def _holders(contract: AgentContract, body: dict[str, Any], model: str) -> list[tuple[Any, Any]]: + """Each (saved form, held object) of one free-text model, wherever the contract holds it. + + The walk follows each field's type, so a model is found in a field of any shape: a list, a + tuple or a Sequence, a dict's or a Mapping's values (a subclass of each, such as an + OrderedDict, too), an Optional or a Union, or inside another model, and an empty one is + passed over. A model kept whole is not looked inside. The contract itself is + ``AgentContract``. + """ + if model == "AgentContract": + return [(body, contract)] + pairs: list[tuple[Any, Any]] = [] + seen: set[int] = set() + + def walk(annotation: Any, saved: Any, held: Any) -> None: + if saved is None or held is None: + return + if isinstance(annotation, type) and issubclass(annotation, BaseModel): + if isinstance(held, BaseModel): + if not isinstance(held, annotation): + return + annotation = type(held) + if annotation.__name__ in KEPT_AS_WRITTEN or not isinstance(saved, dict): + return + if annotation.__name__ == model and id(saved) not in seen: + seen.add(id(saved)) + pairs.append((saved, held)) + for key, field in annotation.model_fields.items(): + inner = held.get(key) if isinstance(held, dict) else getattr(held, key, None) + walk(field.annotation, saved.get(key), inner) + return + origin, arguments = typing.get_origin(annotation), typing.get_args(annotation) + if _walks_as_sequence(origin): + # model_dump keeps a tuple a tuple. A tuple[A, B] has a type for each place. + if arguments and isinstance(saved, (list, tuple)) and isinstance(held, (list, tuple)): + fixed = origin is tuple and not (len(arguments) == 2 and arguments[1] is Ellipsis) + for index, (saved_item, held_item) in enumerate(zip(saved, held)): + if not fixed or index < len(arguments): + walk(arguments[index if fixed else 0], saved_item, held_item) + elif _walks_as_mapping(origin): + if len(arguments) == 2 and isinstance(saved, dict) and isinstance(held, dict): + for key, held_item in held.items(): + walk(arguments[1], saved.get(key), held_item) + else: + # Optional, Union and Annotated: each type it may be. + for argument in arguments: + walk(argument, saved, held) + + walk(type(contract), body, contract) + return pairs + + +def write_contract(contract: AgentContract, destination: Path) -> AgentContract: + """Save the contract to ``destination/contract.json``, its free text cleaned of URL credentials. + + Every writer of the contract goes through here, so no saved contract holds a URL's password, + query or fragment in its prose. The file is written beside its place, synced and moved in, + so a write that fails partway leaves the earlier contract as it was. Only then are the + cleaned values copied back onto ``contract`` itself, in place: the stage and its tools hold + that object, and a fresh one would leave them working from the raw text. It is returned too. + """ + body = contract.model_dump() + places: list[tuple[Any, Any, Any, str]] = [] + for name in FREE_TEXT: + model, field = name.split(".") + for saved, held in _holders(contract, body, model): + if isinstance(saved, dict) and isinstance(saved.get(field), (str, list)): + places.append((saved, held, saved[field], field)) + # One call for the whole save, over a list with no keys: only the URL rule applies, and a + # refused URL is logged once per save rather than once per string. + values = [value for _, _, value, _ in places] + cleaned = _redact(values) + for (saved, _, _, field), value in zip(places, cleaned): + saved[field] = value + + destination = Path(destination) + destination.mkdir(parents=True, exist_ok=True) + path = destination / CONTRACT_FILE + partial = destination / f".{CONTRACT_FILE}.{os.getpid()}.{threading.get_ident()}" + try: + partial.write_text(json.dumps(body, indent=2, ensure_ascii=False), encoding="utf-8") + # Synced before the move, so the move never publishes a file the disk does not hold. + descriptor = os.open(partial, os.O_RDONLY) + try: + os.fsync(descriptor) + finally: + os.close(descriptor) + os.replace(partial, path) + finally: + partial.unlink(missing_ok=True) + + for saved, held, _, field in places: + value = saved[field] + current = held.get(field) if isinstance(held, dict) else getattr(held, field, None) + if isinstance(current, list) and isinstance(value, list): + current[:] = value + elif isinstance(held, dict): + held[field] = value + else: + setattr(held, field, value) + return contract + + def runtime_owns_tools(contract: AgentContract) -> bool: """Whether a submitted voice worker is itself the runnable seam for every tool. diff --git a/src/fi/alk/harness/hosted_chat_entrypoint.py b/src/fi/alk/harness/hosted_chat_entrypoint.py index a1282cc1f..c1a300d22 100644 --- a/src/fi/alk/harness/hosted_chat_entrypoint.py +++ b/src/fi/alk/harness/hosted_chat_entrypoint.py @@ -6,6 +6,7 @@ import traceback import json import os +import re import signal import sys import threading @@ -38,6 +39,14 @@ ) from .tools import schema from .session import ARTIFACT, DONE, RESULT, STAGE_IDLE_TIMEOUT_SECONDS, TEXT, TOOL, Event, Stage +from ..providers.agent import ( + _LETTER, + _QUERY_OR_FRAGMENT, + _REDACTED, + _START, + _URL_BREAKS, + _redact, +) _NUMBERS = {"type": "array", "items": {"type": "integer"}} _RANGES = { @@ -136,6 +145,161 @@ } ) +# The platform stores every chat event as sent, so a URL's password, query and fragment are +# dropped before each emit. Streamed text comes in pieces, and a URL can be split between two, +# so _DeltaCleaner holds back the text after the last cut and cleans each piece it lets out. +# +# The whole message is the reply as stored: the model's text with its leading and trailing +# whitespace stripped. Whitespace before the first character is sent as it comes. +# +# Why a cut is safe: the cleaner never lets a URL cross whitespace, with one exception. A value +# that begins with a scheme and :// and has no whitespace but tabs and line breaks is one URL: +# those are removed and the rest is cleaned as one token. Until the message is known not to be +# such a value, it is held, and every piece let out (by a shed or by the flush) has its tabs and +# line breaks removed before it is cleaned. The message has had no space yet, so each piece is +# one span of that one token, and is cleaned as the whole would clean it. If the message turns +# out to be prose, joining those pieces only removed more: line breaks, and anything a joined +# query or password then owns. +# +# It is known to be prose once its opening (after leading whitespace) is not a scheme's run and +# ://, or once a space has come with more text after it. A space with nothing after it yet +# doesn't decide it: the stripped reply may still be one URL. In prose the whole message is +# cleaned token by token, split at every whitespace; any whitespace is then a cut, and each +# piece is cleaned the prose way too, so it splits at the same tokens and keeps its whitespace. +# +# The holdback is bounded. Past _HOLD_SOFT characters with no cut, what is held is shed: +# - with a ? or # after its first URL start, the rest of the token is that URL's query. The held +# text is sent cleaned (the URL with its host and path), and the rest of the token is dropped +# up to the next cut, which is what cleaning the whole would drop; +# - otherwise everything before its last URL is sent cleaned, and the URL is kept. Each URL +# starts where _redact_token starts it (its scheme's first letter, or its :// when the run +# has no letter), so the pieces are the whole's own pieces; +# - with no URL start, it is sent, keeping only a trailing ":" or ":/" that the next piece could +# turn into "://"; +# - held text that is a whole token and a space after it, before the message is known to be +# prose, is sent cleaned as it is: nothing after the space can join that token. +# A URL still held past _HOLD_HARD characters is sent as [redacted], and the rest of its token is +# dropped. Nothing of a dropped token is ever sent. +_HOLD_SOFT = 4_096 +_HOLD_HARD = 16_384 +_CUT = re.compile(r"[^\S\t\r\n]") +_ANY_CUT = re.compile(r"\s") +_SCHEME_RUN = re.compile(r"[A-Za-z][A-Za-z0-9+.\-]*") +_SCHEME_TAIL = re.compile(r":/?\Z") + + +class _DeltaCleaner: + """The cleaned text of one streamed message, let out piece by piece.""" + + def __init__(self) -> None: + self.held = "" + self.dropping = False + # Whether the whole message is cleaned the prose way, learned from its opening. + self.prose = False + self._opening = "" + # Whether the message's first character has come: whitespace before it is no part of it. + self._begun = False + # Before prose is known: a space has come after the opening began, with only whitespace + # after it so far. + self._spaced = False + + def feed(self, text: str) -> str: + out: list[str] = [] + if not self._begun: + body = text.lstrip() + out.append(text[: len(text) - len(body)]) + text = body + if not text: + return "".join(out) + self._begun = True + was_prose = self.prose + self._learn(text) + if self.prose and not was_prose: + # What was held before may hold tabs, line breaks and a space: it is cut with the rest. + text, self.held = self.held + text, "" + if self.dropping: + cut = (_ANY_CUT if self.prose else _CUT).search(text) + if cut is None: + return "".join(out) + text = text[cut.start() :] + self.dropping = False + last = _ANY_CUT.search(text[::-1]) if self.prose else None + if last is None: + self.held += text + else: + end = len(text) - last.start() + out.append(self._clean(self.held + text[:end])) + self.held = text[end:] + if len(self.held) > _HOLD_SOFT: + out.append(self._shed()) + return "".join(out) + + def flush(self) -> str: + held, self.held, self.dropping = self.held, "", False + return self._clean(held) if held else "" + + def _learn(self, text: str) -> None: + if self.prose or not text: + return + if self._spaced: + if not text.isspace(): + self.prose = True + return + cut = _CUT.search(text) + if cut is not None: + if text[cut.end() :].strip(): + self.prose = True + return + self._spaced = True + if len(self._opening) >= _HOLD_SOFT: + return + self._opening += text[: _HOLD_SOFT - len(self._opening)] + run = _SCHEME_RUN.match(self._opening) + rest = self._opening[run.end() :] if run else self._opening + # Not a letter first, or a scheme's run followed by anything but "://". + if run is None or (rest and not "://".startswith(rest[:3])): + self.prose = True + + def _clean(self, piece: str) -> str: + if not self.prose: + # Joined as cleaning the whole joins a message that is one URL. Whitespace at the end + # is after the message's last character (no part of the stripped reply), so it is + # kept as it came. + body = piece.rstrip() + return _redact(body.translate(_URL_BREAKS)) + piece[len(body) :] + # The space in front only keeps the cleaner from reading the piece as one URL, which + # would drop its tabs and line breaks; the prose way keeps them, and the space comes off. + return _redact(" " + piece)[1:] + + def _shed(self) -> str: + held = self.held + if not self.prose: + if self._spaced: + self.held = "" + return self._clean(held) + self.held = held = held.translate(_URL_BREAKS) + starts = [] + for found in _START.finditer(held): + letter = _LETTER.search(held, found.start(), found.end() - 3) + starts.append(letter.start() if letter else found.end() - 3) + if not starts: + tail = _SCHEME_TAIL.search(held) + keep = tail.start() if tail else len(held) + self.held = held[keep:] + return _redact(held[:keep]) + if _QUERY_OR_FRAGMENT.search(held, starts[0]): + self.held, self.dropping = "", True + return _redact(held) + out = "" + if starts[-1] > 0: + out = _redact(held[: starts[-1]]) + self.held = held = held[starts[-1] :] + if len(held) > _HOLD_HARD: + self.held, self.dropping = "", True + out += _REDACTED + return out + + # Both the foreground coordinator and resumed stage conversations need this boundary: # authoring tools can edit harness artifacts, but cannot update the target configuration # used by the platform or deploy changes to an externally hosted agent. @@ -1654,7 +1818,7 @@ def receive(event: Event) -> None: stage=self.conversation.stage_name, invocation_id=invocation_id, payload={ - "text": "".join(emitted_text).strip(), + "text": _redact("".join(emitted_text).strip()), "interrupted": True, }, ) @@ -1693,19 +1857,24 @@ def receive(event: Event) -> None: return if turn_failures: detail = turn_failures[-1] or "the model turn failed" + # Cleaned whole, stripped as the reply is, before the cut: a cut first could end a URL right + # after a password, and leading whitespace would stop a lone URL being read as one. await self._fail_turn( journal, message_id, assistant_message_id, invocation_id, - f"I couldn't complete that turn: {detail[:400]}", + f"I couldn't complete that turn: {_redact(detail.strip())[:400]}", "stage_failed", ) return - final_text = "".join(emitted_text).strip() + # The model's text is cleaned alone, as the stream cleaned it, and only then are the notes + # added: added first, they would change how it is read (a reply that is one URL broken by + # a line break is cleaned whole). + final_text = _redact("".join(emitted_text).strip()) saved, unsealed = await self._publish_if_changed() if saved: - final_text = f"{final_text}\n\n{saved}".strip() + final_text = f"{final_text}\n\n{_redact(saved)}".strip() if unsealed: # Changes no run could replay are not promoted with this turn. await self._fail_turn( @@ -1723,7 +1892,7 @@ def receive(event: Event) -> None: problems = [line for line in self._last_health.splitlines() if line.strip()] final_text = ( f"{final_text}\n\nStill open from the last health check ({len(problems)}): " - + "; ".join(problems[:3]) + + _redact("; ".join(problems[:3])) + ("; and more" if len(problems) > 3 else "") ).strip() self._last_health = "" @@ -1768,7 +1937,10 @@ async def _fail_turn( text: str, error: str, ) -> None: - """Say why and end the turn failed: its checkpoint is not promoted.""" + """Say why and end the turn failed: its checkpoint is not promoted. + + ``text`` is sent as given, so each caller cleans the model's or the provider's part of it + before composing it.""" await self.client.emit( "assistant_message", message_id=assistant_message_id, @@ -1850,10 +2022,25 @@ async def _pump_events( turn_failures: list[str], ) -> None: visible_stage = self.conversation.stage_name + deltas = _DeltaCleaner() + delta_stage, delta_invocation_id = visible_stage, fallback_invocation_id while True: event = await queue.get() try: if event is None: + # The message ends here, before its assistant_message: what is held goes out. + rest = deltas.flush() + if rest: + try: + await self.client.emit( + "assistant_delta", + message_id=assistant_message_id, + stage=delta_stage, + invocation_id=delta_invocation_id, + payload={"text": rest, "partial": True}, + ) + except Exception: # noqa: BLE001 - the pump still ends. + traceback.print_exc() return if event.kind == DONE and event.detail.get("outcome") == "failed": turn_failures.append( @@ -1875,16 +2062,19 @@ async def _pump_events( visible_stage = stage if event.kind == TEXT and event.text: emitted_text.append(event.text) - await self.client.emit( - "assistant_delta", - message_id=assistant_message_id, - stage=stage, - invocation_id=invocation_id, - payload={ - "text": event.text, - "partial": bool(event.detail.get("partial")), - }, - ) + shown = deltas.feed(event.text) + delta_stage, delta_invocation_id = stage, invocation_id + if shown: + await self.client.emit( + "assistant_delta", + message_id=assistant_message_id, + stage=stage, + invocation_id=invocation_id, + payload={ + "text": shown, + "partial": bool(event.detail.get("partial")), + }, + ) elif event.kind == TOOL: call_id = str(event.detail.get("call_id") or "") tool = event.tool.rsplit("__", 1)[-1] @@ -1924,14 +2114,19 @@ async def _pump_events( stage=stage, invocation_id=invocation_id, function_call_id=call_id or None, - payload={"tool": tool, "text": event.text, "is_error": failed}, + payload={ + "tool": tool, + "text": _redact(event.text), + "is_error": failed, + }, ) elif event.kind == ARTIFACT: await self.client.emit( "tool_result", stage=stage, invocation_id=invocation_id, - payload={"artifact": event.detail.get("path")}, + # Taken from the tool's own words, so it could be a URL. + payload={"artifact": _redact(event.detail.get("path"))}, ) except Exception: # noqa: BLE001 - a lost event or scratch checkpoint must not end the turn. # What the turn says and whether it failed are recorded before any call here. @@ -2018,11 +2213,16 @@ async def _ask( ) temporary.replace(pending_path) await self._checkpoint() + # The model wrote the question and its options. The file above keeps them raw, as asked, so + # a resumed turn still matches them; only the event below is cleaned. The raw file still + # reaches the platform: the workspace checkpoint archives it (TH-9521), or, in + # control-only mode, _checkpoint sends it as coordinator_state instead (TH-9532). + shown_question, *shown_options = _redact([question, *options]) await self.client.emit( "question_requested", message_id=message_id, stage=self.conversation.stage_name, - payload={"prompt": question, "options": options}, + payload={"prompt": shown_question, "options": shown_options}, ) while not self.stopping.is_set(): commands = await self.client.commands() diff --git a/src/fi/alk/harness/hosted_entrypoint.py b/src/fi/alk/harness/hosted_entrypoint.py index 2421a608b..6ccbc8a23 100644 --- a/src/fi/alk/harness/hosted_entrypoint.py +++ b/src/fi/alk/harness/hosted_entrypoint.py @@ -35,6 +35,7 @@ from pathlib import Path from typing import Any, Callable, Protocol, Sequence +from ..providers.agent import _redact from . import observability from . import outbound as ob from .backends import GOOGLE_ADK_APPROVAL_ENV @@ -782,7 +783,8 @@ def _agent_instructions(job: Any, bundle_dir: Path) -> str: parts.append(f"Summary: {summary}") if rules: parts.append("Rules:\n- " + "\n- ".join(rules)) - return "\n\n".join(parts) + # The whole text, once: the job's own prompt, the summary and the rules may each hold a URL. + return _redact("\n\n".join(parts)) def _bundle_contract_value(bundle_dir: Path, key: str) -> str | None: diff --git a/src/fi/alk/harness/job.py b/src/fi/alk/harness/job.py index 7a8a718f1..f35121a71 100644 --- a/src/fi/alk/harness/job.py +++ b/src/fi/alk/harness/job.py @@ -204,6 +204,27 @@ class ProviderExecutionMode(str, Enum): PROVIDER_IMPORT = "provider_import" +PROVIDER_MANAGED_CONNECTORS = frozenset({"vapi", "retell", "retell_chat"}) + + +def provider_profile_source(agent: AgentConnection) -> str | None: + """Where the build loads this agent's profile from, if it loads one. + + A profile from the provider's API or from the job's own config: ``"phone"`` or + ``"livekit"`` for a connect-only job on those connectors, ``"provider"`` for an import or a + connect-only job on a provider-managed connector, and ``None`` when there is none. + """ + connector = agent.connector.strip().lower() + if agent.mode is ProviderExecutionMode.CONNECT_ONLY and connector in {"phone", "livekit"}: + return connector + if agent.mode is ProviderExecutionMode.PROVIDER_IMPORT or ( + agent.mode is ProviderExecutionMode.CONNECT_ONLY + and connector in PROVIDER_MANAGED_CONNECTORS + ): + return "provider" + return None + + class ArtifactLevel(str, Enum): METADATA_ONLY = "metadata-only" TRACES = "traces" diff --git a/src/fi/alk/harness/reading_result.py b/src/fi/alk/harness/reading_result.py index da19fcb5e..e706ee77e 100644 --- a/src/fi/alk/harness/reading_result.py +++ b/src/fi/alk/harness/reading_result.py @@ -18,6 +18,7 @@ from .agent_type_check import EXIT_REFUSED from .backends import StageDone +from .contract import CONTRACT_FILE from .failure_file import FAILURE_FILE from .framework_detection import detect_framework, named_framework_folder from .job import FailureDomain, HarnessJob @@ -37,7 +38,6 @@ logger.addFilter(MaskSecrets()) AGENT_FILE = "agent.json" -CONTRACT_FILE = "contract.json" AGENT_READING_MODEL_ERROR = "agent_reading_model_error" AGENT_READING_INCOMPLETE = "agent_reading_incomplete" diff --git a/src/fi/alk/harness/run/grade.py b/src/fi/alk/harness/run/grade.py index b079f4cf2..1c5668775 100644 --- a/src/fi/alk/harness/run/grade.py +++ b/src/fi/alk/harness/run/grade.py @@ -22,6 +22,7 @@ from dataclasses import dataclass, field from typing import Any +from ...providers.agent import _redact from ..backends import SessionSpec, tool, tool_server from ..catalogue import criteria_text @@ -172,6 +173,8 @@ def _record(scenario: Scenario, transcript: Transcript, ending: str) -> dict[str def _judge_prompt(contract: AgentContract) -> str: + # One call: a contract not saved by write_contract (an older bundle) is cleaned here too. + one_liner, *rules = _redact([contract.one_liner, *contract.hard_constraints[:14]]) return ( "You are grading one run of an agent under test. You are given three kinds of evidence: " "what was said, the actions the agent actually took, and the state of its world " @@ -189,12 +192,8 @@ def _judge_prompt(contract: AgentContract) -> str: "attempted on the way there.\n" " - Declining something holds only if the agent both declined it and gave a true " "reason. Refusing while inventing a reason is not a pass.\n\n" - f"THE AGENT UNDER TEST: {contract.agent} - {contract.one_liner}\n" - + ( - "ITS RULES:\n - " + "\n - ".join(contract.hard_constraints[:14]) - if contract.hard_constraints - else "" - ) + f"THE AGENT UNDER TEST: {contract.agent} - {one_liner}\n" + + ("ITS RULES:\n - " + "\n - ".join(rules) if rules else "") + "\n\nCall submit_verdict once, with one entry per claim, in the order given." ) @@ -242,13 +241,13 @@ def _on_platform( # it says so rather than guessing: the verdict then reads as a failure of the agent when it # was a failure to show the eval the run. record = _record(scenario, transcript, ending) + # Stored on the platform with each eval, so cleaned once here for an older bundle's contract. + rules = _redact(list(contract.hard_constraints)) verdicts: list[Judgement] = [] for claim, name in claims: eval_name = platform_evals.eval_name(contract.agent, name) try: - platform_evals.ensure( - eval_name, claim, contract.agent, contract.hard_constraints - ) + platform_evals.ensure(eval_name, claim, contract.agent, rules) answered = platform_evals.judge(eval_name, record) except Exception as failed: # noqa: BLE001 - one unreachable eval, not a lost suite logging.getLogger(__name__).warning( @@ -299,10 +298,12 @@ def judge_suite_evals( for suite_eval in suite_evals ] verdicts: list[Judgement] = [] + # Sent to the platform, so cleaned once here for an older bundle's contract. + agent_prompt = _redact(contract.system_prompt_excerpt) for suite_eval in suite_evals: inputs = { "conversation": transcript.spoken() or "(nothing was said)", - "agent_prompt": contract.system_prompt_excerpt, + "agent_prompt": agent_prompt, } missing = [name for name in suite_eval.required_inputs if not inputs.get(name)] if missing: diff --git a/src/fi/alk/harness/runtime_repair.py b/src/fi/alk/harness/runtime_repair.py index 370444f06..d22b4eed4 100644 --- a/src/fi/alk/harness/runtime_repair.py +++ b/src/fi/alk/harness/runtime_repair.py @@ -9,9 +9,7 @@ import hashlib import json -import os import re -import tempfile from pathlib import Path from typing import Literal from urllib.parse import urljoin, urlsplit @@ -27,6 +25,7 @@ Runtime, RuntimeConfigFile, generated_chat_interface_problem, + write_contract, ) from .diagnostics import HarnessDiagnostic from .session import DONE, RESULT, TOOL, Event, Stage @@ -1548,20 +1547,8 @@ def apply_runtime_plan_patch( available_secret_aliases=available_secret_aliases, allow_ingress_repair=allow_ingress_repair, ) - contract_path.parent.mkdir(parents=True, exist_ok=True) - descriptor, temporary_name = tempfile.mkstemp( - prefix=".contract-runtime-repair-", suffix=".json", dir=contract_path.parent - ) - temporary = Path(temporary_name) - try: - with os.fdopen(descriptor, "w", encoding="utf-8") as stream: - stream.write(repaired.model_dump_json(indent=2) + "\n") - stream.flush() - os.fsync(stream.fileno()) - os.replace(temporary, contract_path) - finally: - temporary.unlink(missing_ok=True) - return repaired + # write_contract writes beside the file, syncs it and moves it into place. + return write_contract(repaired, contract_path.parent) async def request_runtime_plan_patch( diff --git a/src/fi/alk/harness/sandbox_server.py b/src/fi/alk/harness/sandbox_server.py index cfd3d0586..707b33899 100644 --- a/src/fi/alk/harness/sandbox_server.py +++ b/src/fi/alk/harness/sandbox_server.py @@ -35,6 +35,7 @@ from fi.simulate.runtime.spec import SecretRef +from ..providers.agent import _redact from .credentials import ( CredentialManifest, CredentialRequirement, @@ -58,7 +59,7 @@ ) from .packaging import PackagingManifest, inspect_packaging from .provision import ProvisionError, source_fingerprint, stop -from .secrets import resolve_worker_secrets, worker_environment +from .secrets import SecretResolutionError, resolve_worker_secrets, worker_environment from .source_folder import SubfolderNotFound, agent_folder, subfolder_not_found @@ -1853,14 +1854,19 @@ def _stage_outputs(run_root: Path) -> list[dict[str, Any]]: for tool in contract.get("tools", []) if isinstance(tool, dict) and tool.get("name") ] + # One call: a contract written before write_contract cleaned its prose is cleaned here. + one_liner, *rules = _redact( + [ + str(contract.get("one_liner") or ""), + *(str(item) for item in contract.get("hard_constraints", [])), + ] + ) data = { - "one_liner": str(contract.get("one_liner") or ""), + "one_liner": one_liner, "modality": str(contract.get("modality") or "unknown"), "runtime": contract.get("runtime") or {}, "tools": tools, - "hard_constraints": [ - str(item) for item in contract.get("hard_constraints", []) - ], + "hard_constraints": rules, } outputs.append( { diff --git a/src/fi/alk/harness/scenario_source.py b/src/fi/alk/harness/scenario_source.py index 4d01f76da..b4398a473 100644 --- a/src/fi/alk/harness/scenario_source.py +++ b/src/fi/alk/harness/scenario_source.py @@ -30,7 +30,7 @@ from pydantic import ValidationError from .. import __version__ as _sdk_version -from ..providers.agent import AgentBuildError, build_authored_agent +from ..providers.agent import AgentBuildError, _redact, build_authored_agent from . import outbound as ob from .catalogue import CATALOGUE, SubGoal from .contract import AgentContract @@ -847,7 +847,8 @@ def _chosen_evals_and_prompt(bundle_dir: Path) -> tuple[list[str], str, str]: modality = str(body.get("modality") or "").strip().lower() return ( names, - str(body.get("system_prompt_excerpt") or "").strip(), + # Cleaned again at send, for a contract not saved by write_contract (an older bundle). + _redact(str(body.get("system_prompt_excerpt") or "").strip()), modality if modality in ("voice", "text") else "", ) diff --git a/src/fi/alk/harness/tools.py b/src/fi/alk/harness/tools.py index a86bb63c3..48d734c16 100644 --- a/src/fi/alk/harness/tools.py +++ b/src/fi/alk/harness/tools.py @@ -10,9 +10,7 @@ import asyncio import json import logging -import os import secrets -import threading from pathlib import Path from typing import Any @@ -51,6 +49,8 @@ generated_chat_interface_problem, runtime_public_environment, validate_contract, + write_contract, + CONTRACT_FILE, ) from .failure_file import FAILURE_FILE, write_failure_file from .instrumentation_evidence import detect_instrumentation @@ -419,19 +419,10 @@ def _checked_against_source( def _save_contract(contract: AgentContract, destination: Path) -> dict[str, Any]: - destination.mkdir(parents=True, exist_ok=True) - path = destination / "contract.json" - # Written beside it and moved into place: a write that fails partway leaves the contract an - # earlier call was told was saved as it was. - partial = destination / f".contract.json.{os.getpid()}.{threading.get_ident()}" - try: - partial.write_text( - json.dumps(contract.model_dump(), indent=2, ensure_ascii=False), - encoding="utf-8", - ) - os.replace(partial, path) - finally: - partial.unlink(missing_ok=True) + # write_contract writes beside the file and moves it into place: a write that fails partway + # leaves the contract an earlier call was told was saved as it was. + write_contract(contract, destination) + path = destination / CONTRACT_FILE said = ( f"Accepted and saved to {path}.\n" f"{len(contract.tools)} tools: {', '.join(sorted(contract.tool_names()))}\n" diff --git a/src/fi/alk/providers/agent.py b/src/fi/alk/providers/agent.py index c91490b32..753df1e0b 100644 --- a/src/fi/alk/providers/agent.py +++ b/src/fi/alk/providers/agent.py @@ -1,9 +1,10 @@ from __future__ import annotations import logging +import re from collections.abc import Mapping, Sequence from typing import Any, Literal -from urllib.parse import urlsplit, urlunsplit +from urllib.parse import urlsplit from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator @@ -267,27 +268,104 @@ def _redact_in(value: Any, key: str, refused: list[str]) -> Any: } if isinstance(value, list): return [_redact_in(item, key, refused) for item in value] - if isinstance(value, str) and "://" in value: - try: - parsed = urlsplit(value) - except ValueError as error: - # A netloc urlsplit refuses may still hold credentials, so none of it is kept, and - # only the error's class is logged: its text repeats the netloc. - if not refused: - refused.append(type(error).__name__) - return _REDACTED - if parsed.scheme and parsed.netloc: - hostname = parsed.hostname or "" - try: - port = f":{parsed.port}" if parsed.port else "" - except ValueError: - # A template such as http://{HOST}:{PORT} has no numeric port; - # keep the host rather than failing the whole document. - port = "" - return urlunsplit((parsed.scheme, hostname + port, parsed.path, "", "")) + if isinstance(value, str): + return _redact_text(value, refused) return value +# Every :// starts a URL. Each match is one run of scheme characters before a ://, and the +# lookbehind means a start inside such a run is never tried, which keeps the scan linear. +_START = re.compile(r"(? str: + """Drop every URL's userinfo, query and fragment, and keep the rest of the text as written. + + A URL runs from its scheme to the next whitespace. Within one whitespace-free token, a later + :// that starts before the current URL's own ? or # begins a new URL; once a URL has a ? or + #, the rest of the token is its query or fragment. + """ + if "://" not in value: + return value + if _LEADING_SCHEME.match(value): + joined = value.translate(_URL_BREAKS) + if not _SPACE.search(joined): + # A field that is one URL, perhaps broken by a tab or a line break: one token. + return _redact_token(joined, refused) + return "".join( + part if not part or part.isspace() else _redact_token(part, refused) + for part in _SPACE.split(value) + ) + + +def _redact_token(token: str, refused: list[str]) -> str: + starts: list[tuple[int, bool]] = [] + for match in _START.finditer(token): + # The scheme is the run from its first letter; the text before that letter stays. + letter = _LETTER.search(token, match.start(), match.end() - 3) + starts.append((letter.start() if letter else match.end() - 3, letter is None)) + if not starts: + return token + out = [token[: starts[0][0]]] + # The first ? or # from the first start, searched for once. If it comes before the next + # start, the query owns the rest of the token and the loop ends; if not, it is also the + # first ? or # from every later start up to it. Searching from each start instead would + # make the scan quadratic. + query_at = -1 + for index, (start, schemeless) in enumerate(starts): + if query_at < start: + found = _QUERY_OR_FRAGMENT.search(token, start) + query_at = found.start() if found else len(token) + following = starts[index + 1][0] if index + 1 < len(starts) else len(token) + end = len(token) if query_at < following else following + piece = token[start:end] + if schemeless: + # "${SCHEME}://user:pw@host": parsed with a stand-in scheme, written back without it. + cleaned = _redact_url("x" + piece, refused) + out.append(cleaned[1:] if cleaned.startswith("x://") else cleaned) + else: + out.append(_redact_url(piece, refused)) + if end == len(token): + break + return "".join(out) + + +def _redact_url(url: str, refused: list[str]) -> str: + try: + parsed = urlsplit(url) + except ValueError as error: + # A netloc urlsplit refuses may still hold credentials, so none of it is kept, and + # only the error's class is logged: its text repeats the netloc. + if not refused: + refused.append(type(error).__name__) + return _REDACTED + if not parsed.scheme: + return url + # Built by hand in both branches: urlunsplit drops the // for a scheme urllib does not + # know, so sqlite:///./app.db would come back as sqlite:/./app.db. + if not parsed.netloc: + # file:///x?token=... has no host, but its query and fragment still go. + return f"{parsed.scheme}://{parsed.path}" + hostname = parsed.hostname or "" + if ":" in hostname: + # IPv6 keeps its brackets, so cleaning it a second time changes nothing. + hostname = f"[{hostname}]" + try: + port = f":{parsed.port}" if parsed.port else "" + except ValueError: + # A template such as http://{HOST}:{PORT} has no numeric port; + # keep the host rather than failing the whole document. + port = "" + return f"{parsed.scheme}://{hostname}{port}{parsed.path}" + + def _sorted_directions(listed: Sequence[str]) -> bool: positions = [_DIRECTION_ORDER.index(direction) for direction in listed] return positions == sorted(set(positions)) diff --git a/tests/harness/test_agent_reading.py b/tests/harness/test_agent_reading.py index a3ef8c60b..63e9fd4e0 100644 --- a/tests/harness/test_agent_reading.py +++ b/tests/harness/test_agent_reading.py @@ -10,6 +10,7 @@ import asyncio import builtins import dataclasses +import itertools import json import logging import os @@ -34,12 +35,21 @@ scenario_source, session, spend, + tools, ) from fi.alk.harness.backends import StageDone +from fi.alk.harness.backends.base import qualified from fi.alk.harness.config import chosen_model from fi.alk.harness.contract import AgentContract +from fi.alk.harness.failure_file import write_failure_file from fi.alk.harness.framework_detection import detect_framework -from fi.alk.harness.job import HarnessJob, SourceKind +from fi.alk.harness.job import ( + FailureDomain, + HarnessFailure, + HarnessJob, + HarnessStage, + SourceKind, +) from fi.alk.harness.reading_entry import READING_JOB_INVALID, run_reading_mode from fi.alk.harness.tools import CONTRACT_SERVER, _checked_contract from fi.alk.harness.understand import SUBMIT_REMINDER, reading_settings @@ -534,6 +544,97 @@ def test_a_provider_defined_agent_is_not_read_without_a_build(tmp_path, model, a assert fake.created == [] +# Every connector and mode a job can carry; padded and mixed case as the loader normalises them. +_P_CONNECTORS = [ + "auto", + "http", + "websocket", + "livekit", + "phone", + "vapi", + "retell", + "retell_chat", + " LiveKit ", + "VAPI", + " VAPI ", + "Retell_Chat", +] +_P_MODES = [None, "connect_only", "environment_backed", "provider_import"] +# What the build loads a profile for: from the provider, or from the prompt pasted into the job. +_P_REFUSED = { + ("vapi", "connect_only"), + ("retell", "connect_only"), + ("retell_chat", "connect_only"), + ("phone", "connect_only"), + ("livekit", "connect_only"), + ("vapi", "provider_import"), + ("retell", "provider_import"), + ("retell_chat", "provider_import"), +} + + +def _p_agent(connector: str, mode: str | None) -> dict: + config = { + "assistant_id": "asst-1", + "agent_id": "agent-1", + "agent_name": "lk", + "target_system_prompt": "Be kind.", + } + if connector == "phone": + config["phone_number"] = "+14155550123" + if mode == "environment_backed": + config = {k: v for k, v in config.items() if k not in ("assistant_id", "agent_id")} + return {"connector": connector, "config": config, **({"mode": mode} if mode else {})} + + +def _p_admitted(): + """Every (connector, mode) pair a job validates with.""" + for connector, mode in itertools.product(_P_CONNECTORS, _P_MODES): + try: + HarnessJob.model_validate({**_job(), "agent": _p_agent(connector, mode)}) + except ValueError: + continue + yield connector, mode + + +@pytest.mark.parametrize("connector,mode", list(_p_admitted())) +def test_the_reading_refuses_exactly_what_the_build_loads_a_profile_for( + tmp_path, monkeypatch, connector, mode +): + """Checks the build still uses the shared rule. Both sides call it, so a break inside the rule + moves them together: the refused-pairs table below is the real guard on the rule itself.""" + monkeypatch.setattr( + authoring_entrypoint, "inspect_provider_target", lambda provider, **_: {"provider": provider} + ) + job = HarnessJob.model_validate({**_job(), "agent": _p_agent(connector, mode)}) + secrets = tmp_path / "secrets.json" + secrets.write_text(json.dumps({"VAPI_API_KEY": "x", "RETELL_API_KEY": "y"}), encoding="utf-8") + + assert (agent_reading._check_job(job) is not None) is ( + authoring_entrypoint._load_provider_import_profile(job, secrets) is not None + ) + + +@pytest.mark.parametrize("connector,mode", list(_p_admitted())) +def test_the_refused_jobs_are_the_provider_and_pasted_prompt_ones(connector, mode): + assert _P_REFUSED <= {(c.strip().lower(), m) for c, m in _p_admitted()} + job = HarnessJob.model_validate({**_job(), "agent": _p_agent(connector, mode)}) + + assert (agent_reading._check_job(job) is not None) is ( + (connector.strip().lower(), mode) in _P_REFUSED + ) + + +def test_a_connect_only_job_without_a_provider_is_read(tmp_path, model): + _repo(tmp_path) + fake = model(_writes(CHAT)) + + assert _run(tmp_path, {**_job(), "agent": {"connector": "auto", "mode": "connect_only"}}) == 0 + + assert (tmp_path / "reading" / "agent.json").exists() + assert len(fake.created) == 1 + + def test_a_cancel_before_the_save_is_a_crash_with_its_file(tmp_path, model): _repo(tmp_path) @@ -1539,12 +1640,27 @@ async def saves_when_interrupted(live): def test_nothing_is_saved_after_the_soft_limit_decides(tmp_path, model, monkeypatch): + """A cancel probe: the turn is cancelled and its session closed before the files are read. + + The fake turn waits for the reading's decision itself, not for a time, so it fails only if + the reading reads its files before the AI's turn is cancelled and its session is closed. On + correct code the fake never reaches the gate; the ``stopped`` record is the assertion that + bites. + """ _repo(tmp_path) _short_limit(monkeypatch, soft=0.3) + decided = asyncio.Event() + real_timed_out = agent_reading._timed_out + + def timed_out(output, soft_limit): + decided.set() + return real_timed_out(output, soft_limit) + + monkeypatch.setattr(agent_reading, "_timed_out", timed_out) async def saves_too_late(live): - # Ignores the interrupt, and would save only after the wait has run out. - await asyncio.sleep(1.5) + # Ignores the interrupt, and would save only once the reading has decided. + await decided.wait() await _gate(live, CHAT) yield _done() @@ -1557,6 +1673,320 @@ async def saves_too_late(live): assert fake.stopped == [{"failure": False, "agent": False}] assert not (output / "contract.json").exists() assert _failure_in(output)["code"] == "agent_reading_timed_out" + # Documents the probe: the turn never reached the gate. Not the assertion that bites. + assert fake.submissions == 0 + + +# -- a tool call still running when the reading decides ------------------------------------------ + +CHAT2 = {**CHAT, "agent": "billing", "real_use_cases": ["refund"]} + + +def _replace_the_contract(output: Path) -> None: + _save(output, CHAT2) + + +def _refuse_the_agent(output: Path) -> None: + (output / "contract.json").unlink(missing_ok=True) + write_failure_file( + HarnessFailure( + domain=FailureDomain.AGENT, + stage=HarnessStage.UNDERSTANDING_AGENT, + code="agent_type_unsupported", + message="late refusal", + retryable=False, + ), + output, + ) + + +@pytest.mark.parametrize("path", ["finished", "soft_limit", "close_hangs"]) +@pytest.mark.parametrize( + "late_write", [_replace_the_contract, _refuse_the_agent], ids=["replace", "refuse"] +) +def test_a_late_save_after_the_decision_is_cancelled( + tmp_path, model, monkeypatch, late_write, path +): + """A tool call the session's close did not stop, run as the SDKs run one: a task of its own. + + Its last wait ends while the reading writes agent.json; what it does then is ``late_write``, + a save's own last, synchronous step. The reading cancels it before its next turn. + """ + _repo(tmp_path) + soft = path != "finished" + if soft: + _short_limit(monkeypatch, soft=0.3) + output = tmp_path / "reading" + release = asyncio.Event() + record: dict = {} + real_document = agent_reading.document + + def document(*args, **kwargs): + code = real_document(*args, **kwargs) + release.set() + return code + + monkeypatch.setattr(agent_reading, "document", document) + # classify() calls the document through its own module. + monkeypatch.setattr(reading_result, "document", document) + + async def late(): + try: + await release.wait() + except asyncio.CancelledError: + record["cancelled"] = True + raise + late_write(output) + record["wrote"] = True + + async def turn(live): + _save(live.backend.output, CHAT) + record["task"] = asyncio.get_running_loop().create_task(late()) + if soft: + await asyncio.Event().wait() + yield _done() + + fake = model(turn) + fake.close_hangs = path == "close_hangs" + + assert _run(tmp_path, wait=0.3) == 0 + + assert record.get("cancelled") is True + assert "wrote" not in record + assert json.loads((output / "contract.json").read_text(encoding="utf-8"))["agent"] == "support" + assert (output / "agent.json").exists() + assert not (output / "failure.json").exists() + + +def test_a_late_save_after_a_cancel_that_escapes_the_turn_is_cancelled( + tmp_path, model, monkeypatch +): + """The ending where a cancel escapes the turn after the contract was saved (`_read`'s except + branch). The guard in `finally` covers it too.""" + _repo(tmp_path) + output = tmp_path / "reading" + release = asyncio.Event() + record: dict = {} + real_document = agent_reading.document + + def document(*args, **kwargs): + code = real_document(*args, **kwargs) + release.set() + return code + + monkeypatch.setattr(agent_reading, "document", document) + + async def late(): + try: + await release.wait() + except asyncio.CancelledError: + record["cancelled"] = True + raise + _save(output, CHAT2) + record["wrote"] = True + + async def turn(live): + _save(live.backend.output, CHAT) + record["task"] = asyncio.get_running_loop().create_task(late()) + raise asyncio.CancelledError() + yield _done() + + model(turn) + assert _run(tmp_path) == 0 + assert record.get("cancelled") is True + assert "wrote" not in record + assert json.loads((output / "contract.json").read_text(encoding="utf-8"))["agent"] == "support" + + +def test_a_late_save_woken_in_the_deciding_turn_is_cancelled(tmp_path, model): + """No hook in the document: the late call's wake-up is queued in the same loop turn as the + reading's own, behind it, by a done callback on the conversation's task.""" + _repo(tmp_path) + output = tmp_path / "reading" + record: dict = {} + + async def late(gate): + try: + await gate + except asyncio.CancelledError: + record["cancelled"] = True + raise + _save(output, CHAT2) + record["wrote"] = True + + async def turn(live): + _save(live.backend.output, CHAT) + loop = asyncio.get_running_loop() + gate = loop.create_future() + asyncio.current_task().add_done_callback(lambda _t: gate.done() or gate.set_result(None)) + record["task"] = loop.create_task(late(gate)) + yield _done() + + model(turn) + + assert _run(tmp_path, wait=0.3) == 0 + + assert "wrote" not in record + assert json.loads((output / "contract.json").read_text(encoding="utf-8"))["agent"] == "support" + + +def _reading_spec(tmp_path: Path, job: dict): + """The reading's own stage spec, opened as `_reading` opens it, and its output folder.""" + source = _repo(tmp_path) + output = tmp_path / "reading" + output.mkdir() + harness_job = HarnessJob.model_validate(job) + opened = agent_reading.open_stage( + agent_reading.resolve_source( + "repo", name="agent", root=source, subfolder=harness_job.source.subfolder + ), + out=output, + **reading_settings(harness_job), + )[0] + return opened.spec, output + + +def _submit_contract(spec): + return next(t for t in spec.servers[CONTRACT_SERVER].tools if t.name == "submit_contract") + + +# Saved on its first call: no tools to nudge for, and one hard constraint. +SCAN_PAYLOAD = {k: v for k, v in CHAT.items() if k != "tools"} | { + "hard_constraints": ["Never share another customer's order."] +} + + +@pytest.mark.parametrize("cancelled", [True, False], ids=["cancelled", "uncancelled"]) +@pytest.mark.parametrize("where", ["scan", "checks", "scan-browser"]) +def test_a_cancelled_contract_save_writes_nothing(tmp_path, monkeypatch, where, cancelled): + """The real save, cancelled at each of its waits, writes nothing. + + The refusal path has nothing more to cancel: a refusal needs a second look, which is given + only after the scan, and the refusing call reaches its write with no wait after the scan. + So the scan is its only wait (``scan-browser``). + """ + subfolder = SCRAPER if where == "scan-browser" else SUPPORT + spec, output = _reading_spec(tmp_path, _job(subfolder=subfolder)) + tool = _submit_contract(spec) + entered, release = threading.Event(), threading.Event() + held = "_checked_contract" if where == "checks" else "_tools_in_agent_folder" + real = getattr(tools, held) + + def holding(*args, **kwargs): + entered.set() + assert release.wait(10) + return real(*args, **kwargs) + + monkeypatch.setattr(tools, held, holding) + payload = { + "scan": SCAN_PAYLOAD, + "checks": CHAT, + "scan-browser": {**SCAN_PAYLOAD, "modality": "browser"}, + }[where] + + async def run(): + if where == "checks": + # As the gate sends it: the first, thin call is answered; the second reaches the checks. + first = await tool.handler(dict(CHAT)) + assert not entered.is_set(), first + task = asyncio.ensure_future(tool.handler(dict(payload))) + assert await asyncio.to_thread(entered.wait, 10) + if cancelled: + task.cancel() + # Before the next wait: a save that swallowed the cancel would hold the loop on it. + release.set() + (answer,) = await asyncio.gather(task, return_exceptions=True) + for _ in range(5): + await asyncio.sleep(0) + return task, answer + + task, answer = asyncio.run(run()) + + if cancelled: + assert task.cancelled() + assert not (output / "contract.json").exists() + assert not (output / "failure.json").exists() + elif where == "scan-browser": + assert "second-look" in json.dumps(answer) + assert not (output / "contract.json").exists() + assert not (output / "failure.json").exists() + else: + assert (output / "contract.json").exists(), answer + + +def test_the_readings_gemini_tools_run_on_its_loop(tmp_path, monkeypatch): + """The model calls the real submit_contract; it runs on the reading's own thread and loop. + + So the reading's last cancel reaches it. ADK's tool_thread_pool_config would move it to a + thread's own loop, but only in live mode in the locked 2.7.1, which a reading never uses. + """ + from google.adk.models.google_llm import Gemini + from google.adk.models.llm_response import LlmResponse + from google.adk.runners import Runner + from google.genai import types + + from fi.alk.harness.backends.vertex_gemini import VertexGeminiSession + + calls = [] + + async def generate(self, llm_request, stream=False): + calls.append(llm_request) + if len(calls) == 1: + assert llm_request.contents[-1].parts[0].text == "Hello" + call = types.FunctionCall( + name=qualified(CONTRACT_SERVER, "submit_contract"), args=dict(SCAN_PAYLOAD) + ) + yield LlmResponse( + content=types.Content(role="model", parts=[types.Part(function_call=call)]) + ) + else: + yield LlmResponse(content=types.Content(role="model", parts=[types.Part(text="ok")])) + + monkeypatch.setattr(Gemini, "generate_content_async", generate) + monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") + monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") + monkeypatch.setenv("GOOGLE_GENAI_USE_VERTEXAI", "TRUE") + run_configs = [] + real_run_async = Runner.run_async + + # A plain def: the session wraps what it returns in aclosing(). + def run_async(self, *args, **kwargs): + run_configs.append(kwargs.get("run_config")) + return real_run_async(self, *args, **kwargs) + + monkeypatch.setattr(Runner, "run_async", run_async) + where = [] + real_unwrapped = tools.unwrapped + + # The save's first step; later calls come from its checks, on a worker thread. + def unwrapped(payload): + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + where.append((threading.get_ident(), loop)) + return real_unwrapped(payload) + + monkeypatch.setattr(tools, "unwrapped", unwrapped) + spec, _ = _reading_spec(tmp_path, _job()) + + async def run(): + live = VertexGeminiSession(spec, model="gemini-2.5-flash") + await live.start() + try: + await live.send("Hello") + [event async for event in live.replies()] + finally: + await live.stop() + return threading.get_ident(), asyncio.get_running_loop() + + ident, loop = asyncio.run(run()) + + assert where, "submit_contract was never called" + assert where[0] == (ident, loop) + # A second tripwire, for a later ADK that reads it outside live mode. + assert run_configs and run_configs[0] is not None + assert run_configs[0].tool_thread_pool_config is None def test_a_saved_contract_wins_over_an_error_flag(tmp_path, model): @@ -2163,3 +2593,48 @@ def refuses_the_document(source, destination, *args, **kwargs): # No document, and no half-written one left beside it. assert _names(output) == {"contract.json", "cost.json", "failure.json"} _assert_cost(output, 0.1, True) + + +# -- TH-9491: the canary (plan v5.1 §W7) --------------------------------------------------------- + + +def test_a_password_in_the_prompt_reaches_no_file_the_reading_writes(tmp_path, model): + """The prompt is copied from the source as written, so the URL with credentials is real + input. Every file under the test folder is then searched; only the source holds it.""" + url = "https://bot:hunter2@hooks.example/v1?key=sk-1#frag" + source = _repo(tmp_path) + (source / SUPPORT / "prompt.txt").write_text( + f"You take orders.\nPost each one to {url} after.\n", encoding="utf-8" + ) + model( + _submits( + { + **CHAT, + "one_liner": f"Takes orders and posts them to {url}", + "system_prompt_from": [ + {"path": f"{SUPPORT}/prompt.txt", "start_line": 1, "end_line": 2} + ], + "hard_constraints": [f"Always post to {url}"], + "notes": f"The hook is {url}", + }, + cost_usd=0.1, + ) + ) + output = tmp_path / "reading" + + assert _run(tmp_path, _job(subfolder=SUPPORT)) == 0 + + assert {"contract.json", "agent.json"} <= _names(output) + contract = json.loads((output / "contract.json").read_text(encoding="utf-8")) + assert contract["system_prompt_excerpt"] == ( + "You take orders.\nPost each one to https://hooks.example/v1 after." + ) + holding = sorted( + str(path.relative_to(tmp_path)) + for path in tmp_path.rglob("*") + if path.is_file() + and any( + secret in path.read_bytes() for secret in (b"hunter2", b"sk-1") + ) + ) + assert holding == [f"source/{SUPPORT}/prompt.txt"] diff --git a/tests/harness/test_builder_chat.py b/tests/harness/test_builder_chat.py index bd11669db..703539a07 100644 --- a/tests/harness/test_builder_chat.py +++ b/tests/harness/test_builder_chat.py @@ -2062,3 +2062,648 @@ def declared(path: str, name: str) -> dict: assert sibling["is_error"] and "agents/a" in sibling["content"][0]["text"] assert not own.get("is_error"), own + + +# -- Change review r2, A-H1: what the chat sends the platform loses URL credentials ------------------ + +_SECRET_URL = "https://bot:hunter2@hooks.example/v1?key=sk-1#frag" +_SEEN = "https://hooks.example/v1" +_LEAKS = ("hunter2", "bot:", "sk-1", "key=", "frag", "hun", "ter2") + + +def _leaks(events: list[dict]) -> list[dict]: + return [ + one + for one in events + if any(leak in json.dumps(one.get("payload"), ensure_ascii=False) for leak in _LEAKS) + ] + + +def _deltas(client: BuilderClient) -> list[str]: + return [one["payload"]["text"] for one in client.events if one["kind"] == "assistant_delta"] + + +def test_every_event_the_pump_sends_loses_url_credentials(tmp_path, monkeypatch): + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + queue: asyncio.Queue = asyncio.Queue() + pieces = [ + # One URL split three ways, inside its "://" and inside its password. + "Your agent posts to https:", + "//bot:hun", + "ter2@hooks.example/v1?key=sk-1#frag. It also ", + "reads https://bot:hunter2@hooks.example/v2", + ] + for piece in pieces[:3]: + queue.put_nowait(RuntimeEvent(kind="text", text=piece, detail={"partial": True})) + queue.put_nowait(RuntimeEvent(kind="tool", tool="mcp__x__Read", detail={"call_id": "r"})) + queue.put_nowait( + RuntimeEvent(kind="result", text=f"agent.py:3: HOOK = '{_SECRET_URL}'", detail={"call_id": "r"}) + ) + queue.put_nowait( + RuntimeEvent(kind="artifact", detail={"path": "https://bot:hunter2@hooks.example/out/c.json"}) + ) + queue.put_nowait(RuntimeEvent(kind="text", text=pieces[3], detail={"partial": True})) + queue.put_nowait(None) + said: list[str] = [] + + asyncio.run( + runtime._pump_events( + queue, + assistant_message_id="a-1", + fallback_invocation_id="turn-1", + emitted_text=said, + turn_failures=[], + ) + ) + + assert _leaks(client.events) == [] + assert "".join(_deltas(client)) == ( + f"Your agent posts to {_SEEN} It also reads https://hooks.example/v2" + ) + # The text after the last space is held until the message ends, then sent. + assert client.events[-1]["kind"] == "assistant_delta" + assert client.events[-1]["payload"]["text"] == "https://hooks.example/v2" + results = [one["payload"] for one in client.events if one["kind"] == "tool_result"] + assert results[0]["text"] == f"agent.py:3: HOOK = '{_SEEN}" + assert results[1] == {"artifact": "https://hooks.example/out/c.json"} + # The turn's own record of what was said stays as said; only what is sent is cleaned. + assert said == pieces + + +class _Saying(EditingConversation): + def __init__(self, workspace: Path, events: list[RuntimeEvent]): + super().__init__(workspace, writes=False) + self.events = events + + async def say(self, message, on_event=None): + for event in self.events: + on_event(event) + + +@pytest.mark.parametrize("ending", ["success", "interrupted", "failed"]) +def test_the_turns_assistant_message_loses_url_credentials(tmp_path, monkeypatch, ending): + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime._published_fingerprint() + events = [ + RuntimeEvent(kind="text", text="Saved https://bot:hun", detail={"stage": "builder"}), + RuntimeEvent(kind="text", text="ter2@hooks.example/v1?key=sk-1#frag for you.", detail={}), + ] + if ending == "failed": + events.append( + RuntimeEvent(kind="done", detail={"outcome": "failed", "error": f"provider said {_SECRET_URL}"}) + ) + if ending == "interrupted": + + async def interrupted(self, turn, *, command_sequence): + await turn + return {"kind": "interrupt", "message_id": "m-4", "sequence": 4} + + monkeypatch.setattr(HostedChatRuntime, "_turn_until_interrupted", interrupted) + runtime.conversation = _Saying(tmp_path, events) + + _turn(runtime, {"content": "save it"}) + + assert _leaks(client.events) == [] + assert "".join(_deltas(client)) == f"Saved {_SEEN} for you." + expected = { + "success": f"Saved {_SEEN} for you.", + "interrupted": f"Saved {_SEEN} for you.", + "failed": f"I couldn't complete that turn: provider said {_SEEN}", + }[ending] + assert _reply(client) == expected + + +def test_a_question_the_model_asks_loses_url_credentials(tmp_path, monkeypatch): + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime.stopping.set() + question = f"Should the hook stay {_SECRET_URL} ?" + options = [{"label": "keep", "description": f"post to {_SECRET_URL}"}, "drop it"] + + asyncio.run(runtime._ask("AskUserQuestion", {"question": question, "options": options}, None)) + + asked = next(one for one in client.events if one["kind"] == "question_requested") + assert asked["payload"] == { + "prompt": f"Should the hook stay {_SEEN} ?", + "options": [f"keep: post to {_SEEN}", "drop it"], + } + # The file a resumed turn matches against keeps the question as asked. + pending = json.loads((tmp_path / ".futureagi-pending-question.json").read_text()) + assert pending["question"] == question + + +def test_the_authoring_drain_sends_its_reply_cleaned(monkeypatch): + from fi.alk.harness import cli + + class DrainClient: + command_watermark = 0 + + def __init__(self): + self.events: list[dict] = [] + + async def commands(self): + return [{"sequence": 1, "message_id": "c-1", "payload": {"content": "where?"}}] + + async def emit(self, kind, **kwargs): + self.events.append({"kind": kind, **kwargs}) + + class Stage: + async def say(self, content, on_event): + on_event(RuntimeEvent(kind="text", text=f"It posts to {_SECRET_URL} today.")) + + client = DrainClient() + monkeypatch.delenv("ALK_FOREGROUND_COORDINATOR", raising=False) + monkeypatch.setattr(cli, "_AUTHORING_CLIENT", client) + + asyncio.run(cli._drain_authoring_commands(Stage())) + + reply = next(one for one in client.events if one["kind"] == "assistant_message") + assert reply["payload"]["text"] == f"It posts to {_SEEN} today." + assert _leaks(client.events) == [] + + +# -- The holdback on its own: bounded, and nothing of a URL's credentials let out -------------------- + + +def _stream(text: str, size: int): + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + + cleaner = _DeltaCleaner() + out: list[str] = [] + most = 0 + for start in range(0, len(text), size): + out.append(cleaner.feed(text[start : start + size])) + most = max(most, len(cleaner.held)) + out.append(cleaner.flush()) + return "".join(out), most + + +def test_a_long_run_with_no_space_is_shed_without_its_credentials(): + from fi.alk.harness.hosted_chat_entrypoint import _HOLD_HARD + from fi.alk.providers.agent import _redact + + text = "x" * 50_000 + _SECRET_URL + "y" * 50_000 + " and after." + assert len(text) > 100_000 + + out, most = _stream(text, 7) + + assert most <= _HOLD_HARD + assert "hunter2" not in out and "sk-1" not in out and "bot:" not in out + # The query owns the rest of its token, so the ys go with it, as cleaning the whole drops them. + assert out == _redact(text) == "x" * 50_000 + _SEEN + " and after." + + +def test_a_url_longer_than_the_hard_cap_is_sent_as_redacted(): + from fi.alk.harness.hosted_chat_entrypoint import _HOLD_HARD + + text = "Before https://h.example/" + "p" * _HOLD_HARD + "?key=sk-1&pw=hunter2 after." + + out, most = _stream(text, 100) + + assert most <= _HOLD_HARD + assert out == "Before [redacted] after." + + +def test_prose_around_a_streamed_url_is_kept(): + from fi.alk.providers.agent import _redact + + text = "Read the docs at https://u:hunter2@docs.example/a?x=sk-1 then\tcall us.\n" + + out, _most = _stream(text, 3) + + assert out == _redact(text) == "Read the docs at https://docs.example/a then\tcall us.\n" + + +def test_a_message_that_is_one_url_broken_by_a_line_break_is_cleaned_whole(): + # The cleaner joins a value that is one URL broken by a line break, so no cut is made there. + out, _most = _stream("https://bot:hun\nter2@hooks.example/v1?key=sk-1", 5) + + assert out == _SEEN + + +def test_a_scheme_split_at_its_colon_is_held_over_a_shed(): + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + + cleaner = _DeltaCleaner() + # Past the soft bound with no URL start, so it is shed, all but the ":". + out = cleaner.feed("z" * 5_000 + "https:") + out += cleaner.feed("//bot:hunter2@hooks.example/v1 done") + cleaner.flush() + + assert out == "z" * 5_000 + "https://hooks.example/v1 done" + + +def test_a_shed_inside_a_query_drops_the_rest_of_the_token(): + text = "see https://h.example/cb?a=1&next=https://x.example/y&token=sk-1" + "z" * 5_000 + " ok" + + out, _most = _stream(text, 9) + + assert out == "see https://h.example/cb ok" + + +# -- Change review r2, coordinator items: cuts at any whitespace, and nothing lost past the bound ---- + + +@pytest.mark.parametrize("gap", ["\t", "\n"], ids=["tab", "line_break"]) +def test_a_tab_or_a_line_break_releases_the_held_text(gap): + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + + cleaner = _DeltaCleaner() + + out = cleaner.feed(f"See https://bot:hunter2@hooks.example/v1?key=sk-1{gap}next") + + assert out == f"See {_SEEN}{gap}" + assert cleaner.held == "next" + + +def test_a_message_with_no_space_is_cut_at_a_line_break_once_its_opening_is_prose(): + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + + cleaner = _DeltaCleaner() + + out = cleaner.feed(f"Steps:\n1.{_SECRET_URL}\nnext") + + assert out == f"Steps:\n1.{_SEEN}\n" + assert cleaner.held == "next" + + +def test_a_line_break_after_a_streamed_url_is_kept(): + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + from fi.alk.providers.agent import _redact + + pieces = ["Read\n", "https://u:hunter2@docs.example/a\n", "ok"] + + cleaner = _DeltaCleaner() + out = "".join(cleaner.feed(piece) for piece in pieces) + cleaner.flush() + + assert out == _redact("".join(pieces)) == "Read\nhttps://docs.example/a\nok" + + +@pytest.mark.parametrize( + "text", + [ + "Start https://u:hunter2@h.example/" + "p" * 5_000 + "?key=sk-1" + "q" * 100 + " and the rest.", + "Start " + "x" * 5_000 + "https://u:hunter2@h.example/path and the rest.", + ], + ids=["with_a_query", "after_a_long_word"], +) +def test_past_the_soft_bound_a_held_url_is_sent_cleaned_and_nothing_else_is_lost(text): + """What is shed is the text cleaned, as cleaning the whole would give: the host and path go + out, only the password and query are lost, and the words after the URL are kept.""" + from fi.alk.providers.agent import _redact + + out, _most = _stream(text, 50) + + assert out == _redact(text) + assert "hunter2" not in out and "sk-1" not in out + assert "h.example/" in out and out.endswith(" and the rest.") + + +# -- Change review r3, B-H1 and A-M1/M2: a message that may be one URL broken by a line break --- +# +# The cleaner joins a value that begins with a scheme and :// and has no whitespace but tabs and +# line breaks. The stored reply is the model's text stripped, so the stream must never send what +# cleaning that stripped whole drops, however it is split. + +_PAD = "https://docs.example/" + "a" * 5_000 +_ONE_URL_SHAPES = { + # B-H1: past the soft bound, the kept text starts at a URL start that is not a letter. + "digit_dot_start_tab_in_password": _PAD + "/1.https://bot:hun\tter2@hooks.example/v1?key=sk-1", + "digit_dot_start_break_before_query": ( + _PAD + "/1.https://bot:hunter2@hooks.example/v1\n?key=sk-1" + ), + "schemeless_start_break_in_password": _PAD + "/://bot:hun\nter2@hooks.example/v1?key=sk-1", + "dash_start_break_before_userinfo": _PAD + "/-https://\nbot:hunter2@hooks.example/v1", + # A-M1, shape 2: whitespace before the first character and after the last. + "leading_line_breaks": "\n\nhttps://bot:hun\nter2@hooks.example/v1\n?key=sk-1", + "leading_space": " \thttps://bot:hun\nter2@hooks.example/v1?key=sk-1", + "trailing_spaces": "x://bot:PW1@localho\tst:443/#FS1 ", + "trailing_space_after_a_query": "https://bot:hun\nter2@hooks.example/v1?key=sk-1 \n", + # A-M1, shape 3: an opening run of scheme letters longer than the soft bound. + "long_scheme_run": ( + "a" * 4_097 + "https://U:PW2@hooks.exampl\te/x.json?key=QS1/获取数据x://y" + ), + "long_digit_led_run": "x." * 3_000 + "x://10.\t0.0.1/a/b?key=sk-1", + # A held text that starts at its :// after a shed, and grows past the bound before its query. + "schemeless_held_past_the_bound": ( + "a" * 4_097 + "://bot:hun\tter2@hooks.example/" + "p" * 4_200 + "?key=sk-1" + ), + # Round 2's W1 case, for the record. + "one_url_broken_by_a_line_break": "https://bot:hun\nter2@hooks.example/v1?key=sk-1", + # Fix check, B-M2: the spaced shed of a held lone URL must clean what it lets out. + "lone_url_shed_at_a_space": "https://bot:hun\nter2@h.example/" + "p" * 4_200 + " ", + # Fix check, B-M3: dropping a held query ends only at a space while prose isn't known. + "dropped_query_runs_past_a_line_break": "https://h.example/" + "p" * 4_200 + "?k=v\nQS9tail", +} +_ONE_URL_SECRETS = ( + "hunter2", "hun\t", "hun\n", "ter2", "sk-1", "key=", "bot:", "PW1", "PW2", "FS1", "QS1", + "QS9", +) + + +@pytest.mark.parametrize( + "size", [None, 1, 2, 3, 7, 50], ids=lambda size: f"size_{size or 'whole'}" +) +@pytest.mark.parametrize("name", list(_ONE_URL_SHAPES)) +def test_a_message_that_may_be_one_url_streams_as_its_stripped_whole_is_cleaned(name, size): + from fi.alk.providers.agent import _redact + + text = _ONE_URL_SHAPES[name] + + out, _most = _stream(text, size or len(text)) + + reply = _redact(text.strip()) + assert [secret for secret in _ONE_URL_SECRETS if secret in out] == [] + assert out.strip() == reply + + +@pytest.mark.parametrize( + "text", + [ + # A1: a shed of a run holding two URL starts, the first with userinfo and no query. + "Mirror at https://bot:hunter2@hooks.example/" + + "a" * 5_000 + + "https://cdn.example/x then more", + # A4: the shed with no URL start keeps a trailing ":/". + "Blob " + "b" * 4_100 + "https://bot:hunter2@hooks.example/v1 then more", + ], + ids=["two_starts_in_one_shed", "scheme_tail_held_over_a_shed"], +) +@pytest.mark.parametrize("split", ["size_7", "after_the_scheme_tail"]) +def test_each_shed_sends_what_cleaning_the_whole_sends(text, split): + """Change review r3, A-M2: each must equal the whole, so a one-token plant in _shed fails.""" + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + from fi.alk.providers.agent import _redact + + if split == "size_7": + out, _most = _stream(text, 7) + else: + cut = text.rindex("https:/") + len("https:/") + cleaner = _DeltaCleaner() + out = cleaner.feed(text[:cut]) + cleaner.feed(text[cut:]) + cleaner.flush() + + assert "hunter2" not in out and "bot:" not in out + assert out == _redact(text) + + +def test_a_shed_splits_a_run_where_cleaning_the_whole_splits_it(): + """Each URL starts at its scheme's first letter, as _redact_token starts it, so a digit or + dot before the letter stays with the URL before (here, as part of its port).""" + from fi.alk.providers.agent import _redact + + text = "See https://h.example:80" + "1.ftp://files.example/" + "p" * 4_200 + " then" + + out, _most = _stream(text, 7) + + assert out == _redact(text) == "See https://h.example" + "ftp://files.example/" + "p" * 4_200 + " then" + + +def test_a_shed_finds_a_query_only_after_the_first_url_start(): + """A ? before the URL is not its query, so the shed keeps the URL and the rest of the token + (reviewer A's plant A2 searched from the token's start, and dropped them).""" + from fi.alk.providers.agent import _redact + + text = "Ask q?" + "/" * 100 + "https://h.example/" + "p" * 4_200 + " then" + + out, _most = _stream(text, 7) + + assert out == _redact(text) == text + + +def test_a_space_after_a_long_url_ends_its_token_before_prose_is_known(): + """A space with nothing after it yet doesn't make the reply prose, but nothing after it can + join the token before it, so a shed there must not drop the next word.""" + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + from fi.alk.providers.agent import _redact + + pieces = ["https://bot:hunter2@h.example/" + "p" * 4_200 + "?key=sk-1 ", "then more"] + + cleaner = _DeltaCleaner() + out = "".join(cleaner.feed(piece) for piece in pieces) + cleaner.flush() + + assert out == _redact("".join(pieces)) == "https://h.example/" + "p" * 4_200 + " then more" + + +def test_text_held_before_the_reply_is_known_to_be_prose_is_cut_with_it(): + """What was held while the reply might be one URL keeps its line breaks once it is prose.""" + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + from fi.alk.providers.agent import _redact + + # The long run comes in a delta of its own, so it is held past the soft bound with the text + # before it, and shed. + pieces = ["https://h.example/a\n1.https://h.example/b\n ", "中" * 5_000, " end"] + + cleaner = _DeltaCleaner() + out = "".join(cleaner.feed(piece) for piece in pieces) + cleaner.flush() + + assert out == _redact("".join(pieces)) == "".join(pieces) + + +class _SayingAndWriting(_Saying): + """Says its pieces and leaves one scenario unsaved, so the reply gains the save note.""" + + async def say(self, message, on_event=None): + (self.workspace / "scenarios" / "late_driver").mkdir(parents=True) + (self.workspace / "scenarios" / "late_driver" / "scenario.json").write_text("{}") + await super().say(message, on_event) + + +_BROKEN_URL_PIECES = ["https://bot:hun", "\nter2@hooks.example/v1", "?key=sk-1"] + + +def _texts(pieces: list[str], stage: str = "builder") -> list[RuntimeEvent]: + return [RuntimeEvent(kind="text", text=piece, detail={"stage": stage}) for piece in pieces] + + +def test_the_reply_is_cleaned_before_the_save_note_and_the_health_lines_are_added( + tmp_path, monkeypatch +): + """Change review r3, A-M1 shape 1: notes added first made a one-URL reply prose.""" + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime._published_fingerprint() + monkeypatch.setattr("fi.alk.harness.bundle_author_v2.reseal_scenarios", lambda root: None) + runtime.conversation = _SayingAndWriting(tmp_path, _texts(_BROKEN_URL_PIECES)) + runtime._last_health = f"The hook check failed for {_SECRET_URL}" + + _turn(runtime, {"content": "save it"}) + + assert _leaks(client.events) == [] + assert "".join(_deltas(client)) == _SEEN + assert _reply(client) == ( + f"{_SEEN}\n\nSaved to the environment: added #41 late_driver. It now has 41 scenarios." + f"\n\nStill open from the last health check (1): The hook check failed for {_SEEN}" + ) + + +def test_the_save_note_is_cleaned_on_its_own(tmp_path, monkeypatch): + class NamingClient(BuilderClient): + async def publish(self, digest): + result = await super().publish(digest) + result["added"] = [{"number": 41, "name": f"posts to {_SECRET_URL}"}] + return result + + client = NamingClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime._published_fingerprint() + monkeypatch.setattr("fi.alk.harness.bundle_author_v2.reseal_scenarios", lambda root: None) + runtime.conversation = _SayingAndWriting(tmp_path, _texts(["Added it."])) + + _turn(runtime, {"content": "save it"}) + + assert _leaks(client.events) == [] + # The full stop after #frag is part of the fragment, so it goes too. + assert _reply(client) == ( + f"Added it.\n\nSaved to the environment: added #41 posts to {_SEEN} It now has 41 scenarios." + ) + + +def test_a_failed_reseal_sends_the_reply_cleaned_before_its_note(tmp_path, monkeypatch): + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime._published_fingerprint() + + def fails(root): + raise RuntimeError("seed import failed") + + monkeypatch.setattr("fi.alk.harness.bundle_author_v2.reseal_scenarios", fails) + runtime.conversation = _SayingAndWriting(tmp_path, _texts(_BROKEN_URL_PIECES)) + + _turn(runtime, {"content": "save it"}) + + reply = next(e for e in client.events if e["kind"] == "assistant_message") + assert _leaks(client.events) == [] + assert reply["payload"]["error"] == "reseal_failed" + assert reply["payload"]["text"].startswith(f"{_SEEN}\n\nThese changes were NOT saved") + + +@pytest.mark.parametrize( + ("error", "shown"), + [ + ("https://bot:hun\nter2@hooks.example/v1?key=sk-1", _SEEN), + # Cut first at 400, the password 4242 would be kept as the address's port. + ("x" * 383 + " https://bot:4242@hooks.example/v1", ("x" * 383 + " " + _SEEN)[:400]), + # Fix check B-M1: whitespace around a lone line-broken URL is stripped before cleaning. + ("\n\nhttps://bot:hun\nter2@hooks.example/v1\n?key=sk-1 \n", _SEEN), + ], + ids=[ + "one_url_broken_by_a_line_break", + "cut_right_after_a_numeric_password", + "whitespace_around_one_line_broken_url", + ], +) +def test_a_failed_stage_cleans_its_detail_whole_before_cutting_it( + tmp_path, monkeypatch, error, shown +): + """Change review r3, A-M1 and round 2's L2: the detail is cleaned before the prefix and cut.""" + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime._published_fingerprint() + events = [RuntimeEvent(kind="done", detail={"outcome": "failed", "error": error})] + runtime.conversation = _Saying(tmp_path, events) + + _turn(runtime, {"content": "go"}) + + reply = next(e for e in client.events if e["kind"] == "assistant_message") + assert reply["payload"]["text"] == f"I couldn't complete that turn: {shown}" + assert "4242" not in reply["payload"]["text"] and _leaks(client.events) == [] + + +@pytest.mark.parametrize("ending", ["success", "interrupted"]) +def test_whitespace_around_a_one_url_reply_leaves_the_stream_as_clean_as_the_reply( + tmp_path, monkeypatch, ending +): + """Change review r3, A-M1 shape 2: the reply is stripped, so the stream is read as it is. + Fix check B-M4: the interrupted ending strips its reply too.""" + client = BuilderClient() + runtime = _runtime(tmp_path, client, monkeypatch) + runtime._published_fingerprint() + pieces = ["\n\nhttps://bot:hun", "\nter2@hooks.example/v1", "\n?key=sk-1", " \n"] + if ending == "interrupted": + + async def interrupted(self, turn, *, command_sequence): + await turn + return {"kind": "interrupt", "message_id": "m-4", "sequence": 4} + + monkeypatch.setattr(HostedChatRuntime, "_turn_until_interrupted", interrupted) + runtime.conversation = _Saying(tmp_path, _texts(pieces)) + + _turn(runtime, {"content": "go"}) + + assert _leaks(client.events) == [] + assert "".join(_deltas(client)).strip() == _reply(client) == _SEEN + + +def _fuzzed_message(rng) -> tuple[str, list[str]]: + secrets: list[str] = [] + + def secret(kind: str) -> str: + value = f"{kind}{rng.randrange(10**6, 10**7)}Z" + secrets.append(value) + return value + + def url() -> str: + scheme = rng.choice(["https", "x", "postgres", "1.https", "-https", "", "${S}", "a.b-c"]) + parts = [scheme, "://"] + if rng.random() < 0.7: + parts.append(f"{rng.choice(['bot', secret('UU')])}:{secret('PW')}@") + parts.append(rng.choice(["hooks.example", "h", "[::1]", "10.0.0.1"])) + parts.append(rng.choice(["", "/", "/v1", "/a/b"])) + if rng.random() < 0.5: + parts.append(f"?key={secret('QS')}") + if rng.random() < 0.2: + parts.append(f"#{secret('FS')}") + text = "".join(parts) + if rng.random() < 0.5: + at = rng.randrange(1, len(text)) + text = text[:at] + rng.choice(["\n", "\t", "\r\n"]) + text[at:] + return text + + pieces = [rng.choice(["", "", "\n", " ", "\t\n"])] + if rng.random() < 0.15: + # An opening past the soft bound with no space, to reach every shed. + pieces.append(rng.choice(["a", "x.", "/a", "a\n"]) * rng.randrange(2_100, 2_600)) + for _ in range(rng.randrange(1, 6)): + word = rng.choice(["see", "/", "-", "1.", "://", ","]) + pieces.append(url() if rng.random() < 0.6 else word) + if rng.random() < 0.3: + pieces.append(rng.choice([" ", "\n", "\t", "\xa0"])) + pieces.append(rng.choice(["", "", " ", "\n", " \n"])) + return "".join(pieces), secrets + + +def _random_pieces(text: str, rng) -> list[str]: + cuts = set(rng.sample(range(1, len(text)), min(len(text) - 1, rng.randrange(0, 20)))) + cuts |= { + at + rng.randrange(2) + for at, char in enumerate(text) + if char in ":@?#\n\t" and rng.random() < 0.5 + } + cuts = sorted(at for at in cuts if 0 < at < len(text)) + return [text[start:end] for start, end in zip([0, *cuts], [*cuts, len(text)]) if start < end] + + +def test_random_splits_never_send_a_credential_the_stripped_whole_drops(): + """Change review r3: a bounded random-split fuzz. Whatever the split, no secret leaves the + stream unless cleaning the stripped whole keeps it too (case aside: the whole lowers a + scheme that a shed already let out).""" + import random + + from fi.alk.harness.hosted_chat_entrypoint import _DeltaCleaner + from fi.alk.providers.agent import _redact + + rng = random.Random(9491) + found: list[tuple[str, list[str]]] = [] + for _ in range(600): + text, secrets = _fuzzed_message(rng) + if not text.strip(): + continue + cleaner = _DeltaCleaner() + out = "".join(cleaner.feed(piece) for piece in _random_pieces(text, rng)) + cleaner.flush() + whole = _redact(text.strip()).lower() + shown = out.lower() + leaked = [one for one in secrets if one.lower() in shown and one.lower() not in whole] + if leaked: + found.append((text, leaked)) + assert found == [] diff --git a/tests/harness/test_contract_writer.py b/tests/harness/test_contract_writer.py new file mode 100644 index 000000000..2bcfa3e9a --- /dev/null +++ b/tests/harness/test_contract_writer.py @@ -0,0 +1,1884 @@ +"""The one contract writer (TH-9491, plan v5.1 §2, §3, §V5-V7, §W6-W7). + +Every save of the contract goes through ``write_contract``. It cleans the prose of URL +credentials and keeps the technical fields as written, and it cleans the caller's own object in +place, so the stage that holds it goes on from what was saved. +""" + +from __future__ import annotations + +import ast +import asyncio +import collections +import copy +import inspect +import json +import logging +import os +import re +import subprocess +import sys +import types +import typing +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import pytest +from pydantic import BaseModel, Field + +from fi.alk.harness import ( + amend, + authoring_entrypoint, + authoring_runtime_validation, + builder, + contract as contract_module, + nudge, + runtime_repair, + tools, +) +from fi.alk.harness.contract import ( + FREE_TEXT, + KEPT_AS_WRITTEN, + AgentContract, + Dependency, + Runtime, + RuntimeInterface, + ToolSpec, + write_contract, +) +from fi.alk.providers.agent import _redact + +SOURCE_TREE = Path(contract_module.__file__).resolve().parents[1] +PASSWORD = "hunter2" +KEY = "sk-1" +URL = f"https://bot:{PASSWORD}@hooks.example/v1?key={KEY}#frag" +CLEAN = "https://hooks.example/v1" +PROSE = f"Post to {URL} after each." +CLEANED = f"Post to {CLEAN} after each." + + +def _secret_free(value: Any) -> bool: + text = json.dumps(value, ensure_ascii=False) + return PASSWORD not in text and KEY not in text + + +# -- §3, V7: which fields are cleaned --------------------------------------------------------------- + +# Pinned by name: moving one of these into KEPT_AS_WRITTEN fails here (P16b). +PINNED_FREE_TEXT = ( + "AgentContract.one_liner", + "AgentContract.system_prompt_excerpt", + "AgentContract.hard_constraints", + "AgentContract.implementation", + "AgentContract.real_use_cases", + "AgentContract.caller_identity", + "AgentContract.notes", + "AgentContract.open_questions", + "AgentContract.amendments", + "ToolSpec.description", + "ToolSpec.refuses", + "ToolSpec.effects", + "ToolEntry.notes", + "Dependency.what", +) + + +def _models_in(annotation: Any) -> list[type[BaseModel]]: + """The models a field holds, through list, dict, Optional and Union (W10).""" + found: list[type[BaseModel]] = [] + if inspect.isclass(annotation) and issubclass(annotation, BaseModel): + found.append(annotation) + for argument in typing.get_args(annotation): + found.extend(_models_in(argument)) + return found + + +def _reachable_models() -> dict[str, type[BaseModel]]: + seen: dict[str, type[BaseModel]] = {} + stack: list[type[BaseModel]] = [AgentContract] + while stack: + model = stack.pop() + if model.__name__ in seen: + continue + seen[model.__name__] = model + for field in model.model_fields.values(): + stack.extend(_models_in(field.annotation)) + return seen + + +def _containers_of_models(annotation: Any) -> list[Any]: + """Each generic part of a field's type that holds a model, such as list[ToolSpec].""" + found: list[Any] = [] + if typing.get_origin(annotation) is not None and _models_in(annotation): + found.append(annotation) + for argument in typing.get_args(annotation): + found.extend(_containers_of_models(argument)) + return found + + +def _walked(annotation: Any) -> bool: + """Whether write_contract's walk (contract._holders) follows a type of this shape.""" + origin = typing.get_origin(annotation) + return ( + origin in (typing.Union, types.UnionType, typing.Annotated) + or contract_module._walks_as_sequence(origin) + or contract_module._walks_as_mapping(origin) + ) + + +def test_the_free_text_fields_are_the_fourteen_named() -> None: + assert FREE_TEXT == PINNED_FREE_TEXT + + +def test_every_field_the_contract_reaches_is_classified() -> None: + models = _reachable_models() + # 19 models, three of them from fi.simulate (CompletedToolEvents, CompletedToolFailureEvent, + # CompletedToolRows): a field added there must be classified here too. + assert {"CompletedToolEvents", "CompletedToolFailureEvent", "CompletedToolRows"} <= set(models) + classified = set(FREE_TEXT) | set(KEPT_AS_WRITTEN) + unclassified = [ + f"{name}.{field}" + for name, model in models.items() + if name not in KEPT_AS_WRITTEN + for field in model.model_fields + if f"{name}.{field}" not in classified + ] + assert unclassified == [] + assert not set(FREE_TEXT) & set(KEPT_AS_WRITTEN) + for entry in (*FREE_TEXT, *KEPT_AS_WRITTEN): + model, _, field = entry.partition(".") + assert model in models, entry + assert not field or field in models[model].model_fields, entry + # A model with free text in it is never kept whole. + assert not {entry.split(".")[0] for entry in FREE_TEXT} & set(KEPT_AS_WRITTEN) + # Every field that holds a model holds it in a shape the walk follows, so a new container + # type can't save a model's free text raw (change review r3, B-L1). + unwalked = [ + f"{name}.{field}: {shape}" + for name, model in models.items() + for field, info in model.model_fields.items() + for shape in _containers_of_models(info.annotation) + if not _walked(shape) + ] + assert unwalked == [] + + +@pytest.mark.parametrize( + "order", + [ + ("fi.alk.harness.contract", "fi.alk.providers.agent"), + ("fi.alk.providers.agent", "fi.alk.harness.contract"), + ], +) +def test_the_contract_and_the_cleaner_import_in_either_order(order: tuple[str, str]) -> None: + code = "import importlib\n" + "".join(f"importlib.import_module({name!r})\n" for name in order) + subprocess.run([sys.executable, "-c", code], check=True, timeout=120) + + +# -- B12: what write_contract writes ---------------------------------------------------------------- + +# The KEPT_AS_WRITTEN fields that take a free string, each given a URL with credentials so that +# cleaning one is seen (P10). Literal and pattern-checked fields can't hold one (W10). +KEPT_WITH_A_URL = ( + "agent", + "refusal_signature", + "instrumentation", + "agent_languages", + "chosen_evals", + "data_schema", + "base_environment", + "tools[].arg_values", + "tool_entrypoints[].endpoint", + "tool_entrypoints[].module", + "dependencies[].name", + "dependencies[].engine", + "runtime_dependencies[].name", + "data_store.host", + "data_store.configured_by", +) + + +def _full_contract() -> AgentContract: + """A contract with a URL holding a password and a key in every free-text field, and in each + kept field that takes a free string.""" + body: dict[str, Any] = { + "agent": URL, + "modality": "chat", + "refusal_signature": URL, + "instrumentation": URL, + "agent_languages": [URL], + "chosen_evals": [URL], + "data_schema": {"hook": URL}, + "base_environment": {"hook": [URL]}, + "tools": [ + {"name": "post", "args": ["kind"], "arg_values": {"kind": [URL]}}, + {"name": "book", "args": []}, + ], + "tool_entrypoints": [ + {"tool": "post", "mode": "service", "endpoint": URL, "module": URL}, + ], + "dependencies": [{"name": URL, "engine": URL}], + "runtime_dependencies": [{"name": URL}], + "data_store": {"kind": "database", "host": URL, "configured_by": URL}, + } + for name in FREE_TEXT: + model, field = name.split(".") + holders = ( + [body] + if model == "AgentContract" + else [ + item + for key, kind in ( + ("tools", "ToolSpec"), + ("tool_entrypoints", "ToolEntry"), + ("dependencies", "Dependency"), + ("runtime_dependencies", "Dependency"), + ) + if kind == model + for item in body[key] + ] + ) + for holder in holders: + annotation = AgentContract.model_fields.get(field) + is_list = annotation is not None and typing.get_origin(annotation.annotation) is list + holder[field] = [PROSE, "kept as written"] if is_list else PROSE + return AgentContract.model_validate(body) + + +def _free_values(body: dict[str, Any], name: str) -> list[Any]: + """Every value of one FREE_TEXT name in a saved body.""" + model, field = name.split(".") + if model == "AgentContract": + return [body[field]] + keys = [ + key + for key, info in AgentContract.model_fields.items() + if any(found.__name__ == model for found in _models_in(info.annotation)) + and typing.get_origin(info.annotation) is list + ] + return [item[field] for key in keys for item in body[key]] + + +def _without_free_text(body: dict[str, Any]) -> dict[str, Any]: + body = copy.deepcopy(body) + for name in FREE_TEXT: + model, field = name.split(".") + if model == "AgentContract": + body.pop(field) + continue + for key, info in AgentContract.model_fields.items(): + if any(found.__name__ == model for found in _models_in(info.annotation)): + for item in body.get(key) or []: + item.pop(field) + return body + + +def test_write_contract_cleans_every_free_text_field_and_keeps_every_other_field( + tmp_path: Path, +) -> None: + held = _full_contract() + raw = held.model_dump() + + write_contract(held, tmp_path) + + saved = json.loads((tmp_path / "contract.json").read_text(encoding="utf-8")) + for name in FREE_TEXT: + before, after = _free_values(raw, name), _free_values(saved, name) + assert before and len(before) == len(after), name + for given, cleaned in zip(before, after): + assert cleaned == _redact(given), name + assert _secret_free(cleaned), name + # Every field outside FREE_TEXT is exactly as it was given, URL and all. + assert _without_free_text(saved) == _without_free_text(raw) + kept = json.dumps(_without_free_text(saved)) + assert kept.count(URL) >= len(KEPT_WITH_A_URL) + + +def test_writing_a_written_contract_again_changes_nothing(tmp_path: Path) -> None: + write_contract(_full_contract(), tmp_path / "first") + first = (tmp_path / "first" / "contract.json").read_bytes() + + write_contract(AgentContract.model_validate_json(first), tmp_path / "second") + + assert (tmp_path / "second" / "contract.json").read_bytes() == first + + +def test_write_contract_cleans_the_callers_object_in_place_and_returns_it(tmp_path: Path) -> None: + held = _full_contract() + rules, tool = held.hard_constraints, held.tools[0] + + returned = write_contract(held, tmp_path) + + assert returned is held + assert held.hard_constraints is rules and held.tools[0] is tool + assert held.hard_constraints == [CLEANED, "kept as written"] + assert tool.description == CLEANED + assert held.model_dump() == AgentContract.model_validate_json( + (tmp_path / "contract.json").read_text(encoding="utf-8") + ).model_dump() + + +def test_a_failed_move_leaves_the_saved_contract_and_the_held_one_as_they_were( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + write_contract(AgentContract(agent="desk", one_liner="books desks"), tmp_path) + saved = (tmp_path / "contract.json").read_bytes() + held = AgentContract(agent="desk", one_liner=PROSE) + + def refuse(*_args: Any, **_kwargs: Any) -> None: + raise OSError("no space left on the device") + + monkeypatch.setattr(contract_module.os, "replace", refuse) + with pytest.raises(OSError): + write_contract(held, tmp_path) + + assert (tmp_path / "contract.json").read_bytes() == saved + assert sorted(path.name for path in tmp_path.iterdir()) == ["contract.json"] + # The held object is cleaned only after the move succeeds. + assert held.one_liner == PROSE + + +def test_write_contract_syncs_the_written_file_before_moving_it_in( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Change review r1, A-L2 (plan §2.2): the file moved in is the one synced, synced first.""" + calls: list[tuple[str, int, str]] = [] + real_fsync, real_replace = os.fsync, os.replace + + def fsync(descriptor: int) -> None: + calls.append(("fsync", os.fstat(descriptor).st_ino, "")) + real_fsync(descriptor) + + def replace(source: Any, target: Any) -> None: + calls.append(("replace", os.stat(source).st_ino, Path(target).name)) + real_replace(source, target) + + monkeypatch.setattr(contract_module.os, "fsync", fsync) + monkeypatch.setattr(contract_module.os, "replace", replace) + + write_contract(AgentContract(agent="desk", one_liner=PROSE), tmp_path) + + assert [(kind, name) for kind, _, name in calls] == [("fsync", ""), ("replace", "contract.json")] + assert calls[0][1] == calls[1][1] + assert json.loads((tmp_path / "contract.json").read_text(encoding="utf-8"))["one_liner"] == ( + CLEANED + ) + + +class _Wider(AgentContract): + """A contract with free-text models in field shapes AgentContract doesn't use yet (A-L3).""" + + fallback_tool: ToolSpec | None = None + tools_by_name: dict[str, ToolSpec] = Field(default_factory=dict) + tool_groups: list[list[ToolSpec]] = Field(default_factory=list) + dependency_by_use: dict[str, list[Dependency | None]] = Field(default_factory=dict) + + +def test_write_contract_finds_free_text_models_in_any_field_shape(tmp_path: Path) -> None: + """Change review r1, A-L3: an Optional, a dict's values, nested lists, all cleaned in place.""" + fallback = ToolSpec(name="fallback", description=PROSE, refuses=PROSE) + named = ToolSpec(name="post", description=PROSE, effects=PROSE) + grouped = ToolSpec(name="book", effects=PROSE) + dependency = Dependency(name="hooks", what=PROSE) + held = _Wider( + agent="desk", + fallback_tool=fallback, + tools_by_name={"post": named}, + tool_groups=[[grouped]], + dependency_by_use={"post": [None, dependency]}, + ) + + write_contract(held, tmp_path) + + saved = json.loads((tmp_path / "contract.json").read_text(encoding="utf-8")) + assert _secret_free(saved) + assert saved["fallback_tool"]["description"] == CLEANED + assert saved["fallback_tool"]["refuses"] == CLEANED + assert saved["tools_by_name"]["post"]["description"] == CLEANED + assert saved["tools_by_name"]["post"]["effects"] == CLEANED + assert saved["tool_groups"][0][0]["effects"] == CLEANED + empty, given = saved["dependency_by_use"]["post"] + assert empty is None and given["what"] == CLEANED and given["name"] == "hooks" + # The caller's own objects, in place. + assert held.fallback_tool is fallback and fallback.description == CLEANED + assert held.tools_by_name["post"] is named and named.effects == CLEANED + assert grouped.effects == CLEANED and dependency.what == CLEANED + + +class _Shaped(AgentContract): + """Change review r2, A-L1: tuples, a Sequence and a Mapping, which a first walk missed.""" + + tool_tuple: tuple[ToolSpec, ...] = () + tool_pair: tuple[ToolSpec, Dependency] | None = None + tool_seq: typing.Sequence[ToolSpec] = Field(default_factory=list) + tool_map: typing.Mapping[str, ToolSpec] = Field(default_factory=dict) + tool_list: typing.List[ToolSpec] = Field(default_factory=list) # noqa: UP006 + + +def test_write_contract_finds_free_text_models_in_tuples_sequences_and_mappings( + tmp_path: Path, +) -> None: + """Change review r2, A-L1: each shape is cleaned on disk and on the caller's own objects.""" + tools = {name: ToolSpec(name=name, description=PROSE) for name in ("t", "p", "s", "m", "l")} + dependency = Dependency(name="hooks", what=PROSE) + held = _Shaped( + agent="desk", + tool_tuple=(tools["t"],), + tool_pair=(tools["p"], dependency), + tool_seq=[tools["s"]], + tool_map={"m": tools["m"]}, + tool_list=[tools["l"]], + ) + + write_contract(held, tmp_path) + + saved = json.loads((tmp_path / "contract.json").read_text(encoding="utf-8")) + assert _secret_free(saved) + assert saved["tool_tuple"][0]["description"] == CLEANED + assert saved["tool_pair"][0]["description"] == CLEANED + assert saved["tool_pair"][1]["what"] == CLEANED and saved["tool_pair"][1]["name"] == "hooks" + assert saved["tool_seq"][0]["description"] == CLEANED + assert saved["tool_map"]["m"]["description"] == CLEANED + assert saved["tool_list"][0]["description"] == CLEANED + assert all(tool.description == CLEANED for tool in tools.values()) + assert dependency.what == CLEANED and _secret_free(held.model_dump()) + + +class _Ordered(AgentContract): + """Change review r3, B-L1: a subclass of a mapping, which a membership test missed.""" + + tools_in_order: collections.OrderedDict[str, ToolSpec] = Field( + default_factory=collections.OrderedDict + ) + + +def test_write_contract_finds_free_text_models_in_a_mapping_subclass(tmp_path: Path) -> None: + tool = ToolSpec(name="o", description=PROSE) + held = _Ordered(agent="desk", tools_in_order=collections.OrderedDict(o=tool)) + + write_contract(held, tmp_path) + + saved = json.loads((tmp_path / "contract.json").read_text(encoding="utf-8")) + assert _secret_free(saved) + assert saved["tools_in_order"]["o"]["description"] == CLEANED + assert tool.description == CLEANED + assert _walked(_Ordered.model_fields["tools_in_order"].annotation) + + +def test_write_contract_passes_over_an_empty_optional_model(tmp_path: Path) -> None: + """Change review r1, A-L3: a free-text model field left None saves as None. + + A regression guard only: no plant on the current walk makes it fail, since every branch + already passes a None over. It pins the behaviour for a rewrite of the walk.""" + held = _Wider(agent="desk", one_liner=PROSE) + + write_contract(held, tmp_path) + + saved = json.loads((tmp_path / "contract.json").read_text(encoding="utf-8")) + assert saved["fallback_tool"] is None and saved["one_liner"] == CLEANED + assert held.fallback_tool is None and held.one_liner == CLEANED + + +def test_one_save_logs_one_line_however_many_urls_are_refused( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + refused = "https://admin:hunter2#x@db.example/" + held = AgentContract( + agent="desk", one_liner=refused, notes=refused, hard_constraints=[refused] * 5 + ) + + with caplog.at_level(logging.DEBUG, logger="fi.alk.providers.agent"): + write_contract(held, tmp_path) + + logged = [record for record in caplog.records if record.name == "fi.alk.providers.agent"] + assert [record.getMessage() for record in logged] == [ + "a URL was redacted whole: class=ValueError" + ] + assert PASSWORD not in caplog.text + assert PASSWORD not in (tmp_path / "contract.json").read_text(encoding="utf-8") + + +# -- B13: every writer goes through write_contract -------------------------------------------------- + + +class _Refused(Exception): + pass + + +@pytest.fixture +def refusing_writer(monkeypatch: pytest.MonkeyPatch) -> None: + def refuse(*_args: Any, **_kwargs: Any) -> None: + raise _Refused + + # Every module that holds the name, so no caller can reach the real one. + for module in (contract_module, tools, amend, runtime_repair): + monkeypatch.setattr(module, "write_contract", refuse) + + +def _amendable() -> AgentContract: + return AgentContract( + agent="desk", + one_liner="books desks", + real_use_cases=["book a desk"], + hard_constraints=["Confirm before booking"], + tools=[{"name": "post", "args": ["kind"], "arg_values": {"kind": ["a"]}}], + tool_entrypoints=[{"tool": "gone", "mode": "unreachable"}], + ) + + +def _declare_tool(tmp_path: Path, contract: AgentContract, description: str = "cancel") -> Any: + source = tmp_path / "source" + source.mkdir(exist_ok=True) + (source / "agent.py").write_text("def cancel():\n return 'ok'\n", encoding="utf-8") + server = builder.suite_server(tmp_path / "out", [], contract, source) + declare = next(one for one in server.tools if one.name == "declare_tool") + return asyncio.run( + declare.handler( + { + "name": "cancel", + "description": description, + "evidence_file": "agent.py", + "evidence_quote": "def cancel():", + "why": "the agent has it", + } + ) + ) + + +def _runtime_repair(tmp_path: Path) -> None: + source = tmp_path / "source" + service = source / "service" + service.mkdir(parents=True) + (service / "pyproject.toml").write_text("[project]\nname='agent'\nversion='1'\n") + (service / "app.py").write_text("# entrypoint\n") + contract = AgentContract(agent="support-agent", runtime=Runtime(language="python")) + contract_path = tmp_path / "contract.json" + contract_path.write_text(contract.model_dump_json(indent=2) + "\n") + runtime_repair.apply_runtime_plan_patch( + source, + contract_path, + runtime_repair.RuntimePlanPatch( + base_contract_hash=runtime_repair._contract_hash(contract), + runtime=Runtime( + language="python", + workdir="service", + command=["python", "app.py"], + interface=RuntimeInterface(kind="http", port=8080, path="/chat"), + ), + evidence_paths=("service/pyproject.toml", "service/app.py"), + diagnostic_codes=("generated_runtime_plan_invalid",), + summary="Run the HTTP service declared by the submitted component.", + ), + allowed_diagnostic_codes={"generated_runtime_plan_invalid"}, + ) + + +def _data_store_repair(tmp_path: Path) -> int: + """V8 row 5: the data-store repair's helper, called directly with a fake candidate folder.""" + authoring = tmp_path / "authoring" + authoring.mkdir() + contract = AgentContract(agent="desk", real_use_cases=["book"]) + (authoring / "contract.json").write_text(contract.model_dump_json(), encoding="utf-8") + candidate = tmp_path / "candidate" + candidate.mkdir() + (candidate / "contract.json").write_text( + AgentContract(agent="desk", data_store={"kind": "in_memory"}).model_dump_json(), + encoding="utf-8", + ) + return authoring_runtime_validation._promote_data_store(contract, authoring, candidate) + + +def _write_stale_contract(tmp_path: Path) -> Path: + workspace = tmp_path / "workspace" + workspace.mkdir() + (workspace / "contract.json").write_text(_amendable().model_dump_json(), encoding="utf-8") + return workspace + + +_WRITERS = { + "_save_contract": lambda tmp: tools._save_contract(_amendable(), tmp), + "widen": lambda tmp: amend.widen( + _amendable(), tmp, tool_name="post", argument="kind", values=["b"], why="w" + ), + "add_rule": lambda tmp: amend.add_rule(_amendable(), tmp, rule="Be brief", why="w"), + "drop_rule": lambda tmp: amend.drop_rule( + _amendable(), tmp, rule="Confirm before booking", why="w" + ), + "fix_tool": lambda tmp: amend.fix_tool( + _amendable(), tmp, tool_name="post", why="w", description="posts" + ), + "set_modality": lambda tmp: amend.set_modality(_amendable(), tmp, modality="voice", why="w"), + "builder.declare_tool": lambda tmp: _declare_tool(tmp, _amendable()), + "nudge": lambda tmp: nudge.drop_stale_tool_references(_write_stale_contract(tmp)), + "authoring_entrypoint": lambda tmp: authoring_entrypoint._refresh_source_facts( + None, tmp, _write_stale_contract(tmp) + ), + "runtime_plan_repair": _runtime_repair, + "data_store_repair": _data_store_repair, +} + + +@pytest.mark.parametrize("writer", list(_WRITERS)) +def test_every_contract_writer_goes_through_write_contract( + writer: str, tmp_path: Path, refusing_writer: None +) -> None: + with pytest.raises(_Refused): + _WRITERS[writer](tmp_path) + + +# The static half of B13 (W7). A function is flagged when it names the contract file and calls a +# write verb; the allow-list holds every such function by (module, function, verb, count), so a +# new write in any of them, or a new function, fails the test. The scan reads the source, so it +# is a heuristic: the raising patch above and the canaries are what prove each writer goes +# through write_contract. Change review r1 (A-M1) widened it to the nine shapes it missed: +# a path held on self, helpers to any depth, methods and nested defs as helpers, lambdas, an +# aliased CONTRACT_FILE, functions imported from shutil and os, the file named at the end of a +# path, an open mode held in a variable, and os.open, os.fdopen and os.link. Change review r2 +# (A-M1) added three more: a module-level name bound to the file, imported into another module +# under any name or read as module.NAME; a helper method called on an instance of a class +# defined in the same module (Keeper().save(...), or keeper = Keeper() then keeper.save(...)); +# and zipfile's writestr and tarfile's addfile. Change review r3 (B-M1) added three more: a static +# or class method called on a class defined in the same module (Store.keep(...)); a class +# constant bound to the file in a class body, read as Names.CONTRACT_JSON; and a name bound to a +# marker by an import (from .contract import CONTRACT_FILE as CONTRACT_JSON), which is a marker +# in every module that imports it from there. +# +# What it still can't see, each a way to write the file on purpose rather than by accident: +# - a getter that returns the path (contract_path(out).write_text(...)); +# - a path handed to a constructor and written by a method of the object +# (JsonFile(out / "contract.json").save(body), B's plant R3a); +# - a copy run as a process (subprocess and cp), or through functools.partial, getattr or +# io.FileIO; +# - a name built at run time ("contract" + ".json"), the file named as bytes (b"contract.json"), +# or a module global set in one function and written in another; +# - a helper kept in a module-level lambda (_keep = lambda path, body: path.write_text(body)), +# or a bound method kept in a name (write = (out / "contract.json").write_text; write(body)); +# - a def that hands its own parameter to an instance method called on its class with the +# instance passed by hand (Keeper.save(keeper, path)) is not made a helper, since the slots +# are read one place off; the call itself is flagged where the file is named; +# - a class constant read on a class from another module, or on an instance other than self; +# - ZipFile.write and TarFile.add with the file as the archive name, since write and add are +# too common to be verbs; +# - a helper method inherited from a class in another module, a method called on an instance +# of an imported class, and a helper function re-exported through a package __init__; +# - a changed target: the allow-list keys each function by (module, function, verb, count), so +# a write already counted in an allow-listed function can be pointed at contract.json without +# changing its count, and the scan still passes. +MARKER_NAMES = frozenset({"CONTRACT", "CONTRACT_FILE"}) +MARKER_TEXT = "contract.json" +# The file by itself, or as the last part of a path: "authoring/contract.json". +_MARKER_PATH = re.compile(r"(?:^|[/\\])" + re.escape(MARKER_TEXT) + r"$") +_DOTTED_VERBS = frozenset( + { + "os.replace", + "os.rename", + "os.renames", + "os.link", + "os.symlink", + "shutil.copy", + "shutil.copy2", + "shutil.copyfile", + "shutil.copytree", + "shutil.move", + "json.dump", + } +) +# A call into one of these modules is a verb only when it is listed above or is an open that +# writes, never by the attribute rules below (os.open is not Path.open). +_KNOWN_MODULES = frozenset({"os", "shutil", "json", "io", "codecs"}) +_WRITE_FLAGS = frozenset({"O_WRONLY", "O_RDWR", "O_CREAT", "O_APPEND", "O_TRUNC", "O_EXCL"}) +_READ_FLAGS = frozenset( + {"os", "O_RDONLY", "O_CLOEXEC", "O_NOFOLLOW", "O_BINARY", "O_NONBLOCK", "O_DIRECTORY"} +) +_SELF = frozenset({"self", "cls"}) +_DEFS = (ast.FunctionDef, ast.AsyncFunctionDef) +# A lambda is not a scope of its own: its body counts in the def around it. +_SCOPES = (*_DEFS, ast.ClassDef) + + +@dataclass(frozen=True) +class _Module: + """What a module's names resolve to, for telling which def a call reaches.""" + + name: str + # Local names bound to a module: the stdlib's, or one of this package's. + modules: dict[str, str] + # Local names bound to a function of _KNOWN_MODULES, as "module.function". + functions: dict[str, str] + # Local names imported from one of this package's modules, as (module, name). + names: dict[str, tuple[str, str]] + # The functions defined here, at any depth (methods aside). + defined: frozenset[str] + # The classes defined here, at any depth, and the names bound anywhere here to an instance + # of one: keeper = Keeper(...). + classes: frozenset[str] = frozenset() + instances: frozenset[str] = frozenset() + + +def _module_name(path: Path, root: Path) -> str: + parts = path.relative_to(root.parent.parent).with_suffix("").parts + return ".".join(parts[:-1] if parts[-1] == "__init__" else parts) + + +def _context(tree: ast.Module, path: Path, root: Path, package: frozenset[str]) -> _Module: + name = _module_name(path, root) + here = name.split(".") if path.name == "__init__.py" else name.split(".")[:-1] + modules = {one: one for one in _KNOWN_MODULES} + functions: dict[str, str] = {} + names: dict[str, tuple[str, str]] = {} + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + if alias.asname: + modules[alias.asname] = alias.name + else: + modules[alias.name.split(".")[0]] = alias.name.split(".")[0] + elif isinstance(node, ast.ImportFrom): + base = here[: len(here) - node.level + 1] if node.level else [] + source = ".".join([*base, *([node.module] if node.module else [])]) + for alias in node.names: + local = alias.asname or alias.name + if f"{source}.{alias.name}" in package: + modules[local] = f"{source}.{alias.name}" + elif source in _KNOWN_MODULES: + functions[local] = f"{source}.{alias.name}" + else: + names[local] = (source, alias.name) + defined = frozenset(node.name for node, method in _defs(tree) if not method) + classes = frozenset(node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef)) + instances: set[str] = set() + for node in ast.walk(tree): + if isinstance(node, ast.Assign): + value, targets = node.value, node.targets + elif isinstance(node, (ast.AnnAssign, ast.NamedExpr)) and node.value is not None: + value, targets = node.value, [node.target] + elif isinstance(node, ast.withitem) and node.optional_vars is not None: + value, targets = node.context_expr, [node.optional_vars] + else: + continue + if _instance_of(value, classes): + instances |= {one.id for one in targets if isinstance(one, ast.Name)} + return _Module(name, modules, functions, names, defined, classes, frozenset(instances)) + + +def _instance_of(node: ast.AST, classes: frozenset[str]) -> bool: + """Whether a node makes an instance of one of these classes: Keeper(...).""" + return ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id in classes + ) + + +def _marker_aliases(tree: ast.Module) -> frozenset[str]: + """Names CONTRACT or CONTRACT_FILE is imported under: ``import CONTRACT_FILE as _NAME``.""" + return frozenset( + alias.asname + for node in ast.walk(tree) + if isinstance(node, ast.ImportFrom) + for alias in node.names + if alias.name in MARKER_NAMES and alias.asname + ) + + +def _docstrings(tree: ast.AST) -> set[int]: + found = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)): + body = node.body + if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant): + found.add(id(body[0].value)) + return found + + +def _own_nodes(scope: ast.AST) -> list[ast.AST]: + """A scope's own nodes: a nested def or class is scanned as its own scope, a lambda is not.""" + nodes: list[ast.AST] = [] + stack = list(ast.iter_child_nodes(scope)) + while stack: + node = stack.pop() + nodes.append(node) + if not isinstance(node, _SCOPES): + stack.extend(ast.iter_child_nodes(node)) + return nodes + + +def _argument(call: ast.Call, position: int, keyword: str) -> ast.AST | None: + for one in call.keywords: + if one.arg == keyword: + return one.value + return call.args[position] if len(call.args) > position else None + + +def _mode_writes(mode: ast.AST | None) -> bool: + """Whether an open mode may write. One held in a variable may.""" + if mode is None: + return False + if isinstance(mode, ast.Constant): + return isinstance(mode.value, str) and any(flag in mode.value for flag in "wax+") + return True + + +def _flags_write(call: ast.Call) -> bool: + """Whether os.open's flags may write: a write flag, or flags this can't read.""" + flags = _argument(call, 1, "flags") + if flags is None: + return False + if isinstance(flags, ast.Constant): + return flags.value != 0 + names = {sub.attr for sub in ast.walk(flags) if isinstance(sub, ast.Attribute)} + names |= _names_in(flags) + return bool(names & _WRITE_FLAGS or names - _READ_FLAGS - _WRITE_FLAGS) + + +def _dotted(func: ast.AST, imports: _Module) -> str | None: + """``module.function`` for a call into one of _KNOWN_MODULES, through any import alias.""" + if isinstance(func, ast.Name): + return imports.functions.get(func.id) + if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name): + module = imports.modules.get(func.value.id) + if module in _KNOWN_MODULES: + return f"{module}.{func.attr}" + return None + + +def _written(call: ast.Call, slots: frozenset[int | str]) -> list[ast.AST]: + """The arguments a call to a helper writes to: those in its slots, and any * or ** ones.""" + targets: list[ast.AST] = [ + argument + for index, argument in enumerate(call.args) + if index in slots or isinstance(argument, ast.Starred) + ] + targets += [one.value for one in call.keywords if one.arg is None or one.arg in slots] + return targets + + +# A helper is keyed by ("function" or "method", module, name). A call reaches one through the +# module's own names: a bare name defined here or imported from where it is defined, a function +# on an imported module, or a method of this module on self or cls or on an instance of a class +# defined here. A call on any other object is never a helper call, so a common name such as get +# or append can't spread through the tree. +_Helpers = dict[tuple[str, str, str], frozenset[int | str]] + + +def _helper(func: ast.AST, helpers: _Helpers, imports: _Module) -> frozenset[int | str] | None: + if isinstance(func, ast.Name): + if func.id in imports.defined: + return helpers.get(("function", imports.name, func.id)) + if func.id in imports.names: + return helpers.get(("function", *imports.names[func.id])) + return None + if isinstance(func, ast.Attribute) and _instance_of(func.value, imports.classes): + return helpers.get(("method", imports.name, func.attr)) + if isinstance(func, ast.Attribute) and isinstance(func.value, ast.Name): + if func.value.id in _SELF or func.value.id in imports.instances: + return helpers.get(("method", imports.name, func.attr)) + if func.value.id in imports.classes: + # A static or class method called on its class: Store.keep(...) (change review r3). + return helpers.get(("method", imports.name, func.attr)) + module = imports.modules.get(func.value.id) + if module is not None: + return helpers.get(("function", module, func.attr)) + return None + + +def _verb(call: ast.Call, helpers: _Helpers, imports: _Module) -> tuple[str, list[ast.AST]] | None: + """The write verb a call is, with the expressions it writes to.""" + func = call.func + dotted = _dotted(func, imports) + if dotted in _DOTTED_VERBS: + return dotted, list(call.args[:2]) + if dotted == "os.open": + return (dotted, list(call.args[:1])) if _flags_write(call) else None + if dotted in ("os.fdopen", "io.open", "codecs.open"): + return (dotted, list(call.args[:1])) if _mode_writes(_argument(call, 1, "mode")) else None + if dotted is not None: + return None + if isinstance(func, ast.Attribute): + if func.attr in ("write_text", "write_bytes"): + return func.attr, [func.value] + # Path.replace(target) and Path.rename(target) take one argument; str.replace two. + if func.attr in ("replace", "rename") and len(call.args) == 1 and not call.keywords: + return f"Path.{func.attr}", [func.value, call.args[0]] + if func.attr in ("hardlink_to", "symlink_to") and len(call.args) == 1: + return f"Path.{func.attr}", [func.value, call.args[0]] + if func.attr == "open": + # tarfile.open(name, mode) takes the mode second; Path.open(mode) first. + on_a_module = isinstance(func.value, ast.Name) and func.value.id in imports.modules + mode = _argument(call, 1 if on_a_module else 0, "mode") + return ("Path.open", [func.value]) if _mode_writes(mode) else None + if func.attr == "_write_whole": + return func.attr, list(call.args[:1]) + if func.attr in ("writestr", "addfile"): + # ZipFile.writestr(arcname, data) and TarFile.addfile(tarinfo, fileobj). + return func.attr, list(call.args[:1]) + slots = _helper(func, helpers, imports) + return (func.attr, _written(call, slots)) if slots is not None else None + if isinstance(func, ast.Name): + if func.id == "open": + return ("open", list(call.args[:1])) if _mode_writes(_argument(call, 1, "mode")) else None + if func.id == "_write_whole": + return func.id, list(call.args[:1]) + slots = _helper(func, helpers, imports) + return (func.id, _written(call, slots)) if slots is not None else None + return None + + +def _names_in(node: ast.AST) -> set[str]: + return {sub.id for sub in ast.walk(node) if isinstance(sub, ast.Name)} + + +def _bound(nodes: list[ast.AST], seeds: set[str], is_seed: Any = None) -> set[str]: + """The seeds, and every name bound here from an expression holding one, to a fixed point.""" + marked = set(seeds) + pairs: list[tuple[set[str], bool, set[str]]] = [] + for node in nodes: + if isinstance(node, ast.Assign): + value, targets = node.value, node.targets + elif isinstance(node, (ast.AnnAssign, ast.AugAssign, ast.NamedExpr)) and node.value: + value, targets = node.value, [node.target] + elif isinstance(node, ast.withitem) and node.optional_vars is not None: + value, targets = node.context_expr, [node.optional_vars] + elif isinstance(node, (ast.For, ast.AsyncFor)): + value, targets = node.iter, [node.target] + else: + continue + seeded = is_seed is not None and any(is_seed(sub) for sub in ast.walk(value)) + pairs.append((_names_in(value), seeded, set().union(*map(_names_in, targets)))) + changed = True + while changed: + changed = False + for names, seeded, bound in pairs: + if (seeded or names & marked) and bound - marked: + marked |= bound + changed = True + return marked + + +def _defs(tree: ast.Module) -> list[tuple[ast.FunctionDef | ast.AsyncFunctionDef, bool]]: + """Every def in a module, at any depth, and whether it is a method (a def nested in a method + is a function: it is called by its name).""" + found: list[tuple[ast.FunctionDef | ast.AsyncFunctionDef, bool]] = [] + + def collect(node: ast.AST, in_class: bool) -> None: + for child in ast.iter_child_nodes(node): + if isinstance(child, _DEFS): + found.append((child, in_class)) + collect(child, False) + elif isinstance(child, ast.ClassDef): + collect(child, True) + else: + collect(child, in_class) + + collect(tree, False) + return found + + +def _parameters(node: ast.FunctionDef | ast.AsyncFunctionDef, method: bool) -> list[tuple[int | None, str]]: + """A def's parameters by (position, name); a method's self or cls is not one of them.""" + arguments = node.args + positional = [one.arg for one in (*arguments.posonlyargs, *arguments.args)] + static = any( + isinstance(decorator, ast.Name) and decorator.id == "staticmethod" + for decorator in node.decorator_list + ) + if method and not static and positional and positional[0] in _SELF: + positional = positional[1:] + named: list[tuple[int | None, str]] = list(enumerate(positional)) + named += [(None, one.arg) for one in arguments.kwonlyargs] + return named + + +def _writer_helpers(trees: dict[Path, ast.Module], imports: dict[Path, _Module]) -> _Helpers: + """Functions, methods and nested defs that write to one of their parameters, each with the + parameters it writes to (by position and by name). Found to a fixed point: a def that hands + a parameter to a known helper is a helper too, however deep. + + write_contract is the sanctioned writer, so a call to it is never a write verb here.""" + prepared = [] + for path, tree in trees.items(): + for node, method in _defs(tree): + if node.name == "write_contract": + continue + own = _own_nodes(node) + calls = [sub for sub in own if isinstance(sub, ast.Call)] + if not calls: + continue + derived = [ + (slot, name, _bound(own, {name})) for slot, name in _parameters(node, method) + ] + key = ("method" if method else "function", imports[path].name, node.name) + prepared.append((key, calls, derived, imports[path])) + helpers: dict[tuple[str, str, str], set[int | str]] = {} + changed = True + while changed: + changed = False + known = {key: frozenset(slots) for key, slots in helpers.items()} + for key, calls, derived, module_imports in prepared: + for call in calls: + found = _verb(call, known, module_imports) + if not found: + continue + written = set().union(*map(_names_in, found[1])) + for slot, parameter, names in derived: + if written & names: + fresh = {parameter} if slot is None else {parameter, slot} + slots = helpers.setdefault(key, set()) + if not fresh <= slots: + slots |= fresh + changed = True + return {key: frozenset(slots) for key, slots in helpers.items()} + + +def _self_attrs(node: ast.ClassDef, marked: set[str], is_marker: Any) -> frozenset[str]: + """The attributes of self or cls that hold the contract file in a class, set in any of its + methods or in its body, to a fixed point.""" + scopes = [_own_nodes(node)] + [_own_nodes(sub) for sub in ast.walk(node) if isinstance(sub, _DEFS)] + attrs: set[str] = set() + while True: + known = frozenset(attrs) + + def seed(sub: ast.AST, known: frozenset[str] = known) -> bool: + return is_marker(sub, known) + + for index, own in enumerate(scopes): + names = _bound(own, marked, seed) - _SELF + for sub in own: + if isinstance(sub, ast.Assign): + value, targets = sub.value, sub.targets + elif isinstance(sub, (ast.AnnAssign, ast.AugAssign)) and sub.value is not None: + value, targets = sub.value, [sub.target] + else: + continue + if not (_names_in(value) & names or any(seed(part) for part in ast.walk(value))): + continue + for target in targets: + for part in ast.walk(target): + if isinstance(part, ast.Attribute) and isinstance(part.value, ast.Name): + if part.value.id in _SELF: + attrs.add(part.attr) + elif isinstance(part, ast.Name) and index == 0: + # A class constant, read as self.NAME or cls.NAME. + attrs.add(part.id) + if attrs == known: + return known + + +def _names_the_file( + sub: ast.AST, + docs: set[int], + marker_names: frozenset[str], + imports: _Module, + exported: dict[str, frozenset[str]], +) -> bool: + """Whether a node names the contract file: the text, a marker name, or module.NAME for a + name another module binds to the file.""" + if isinstance(sub, ast.Constant): + return isinstance(sub.value, str) and bool(_MARKER_PATH.search(sub.value)) and id(sub) not in docs + if isinstance(sub, ast.Name): + return sub.id in marker_names + if isinstance(sub, ast.Attribute): + if sub.attr in MARKER_NAMES: + return True + if isinstance(sub.value, ast.Name) and sub.value.id in imports.modules: + return sub.attr in exported.get(imports.modules[sub.value.id], frozenset()) + return False + + +def _marker_names( + tree: ast.Module, imports: _Module, exported: dict[str, frozenset[str]] +) -> frozenset[str]: + """The names a module reads the contract file by: CONTRACT and CONTRACT_FILE under any name, + and any name imported from a module that binds it to the file.""" + imported = { + local + for local, (source, name) in imports.names.items() + if name in exported.get(source, frozenset()) + } + return MARKER_NAMES | _marker_aliases(tree) | imported + + +def _exported_markers( + trees: dict[Path, ast.Module], imports: dict[Path, _Module] +) -> dict[str, frozenset[str]]: + """Each module's module-level names bound to the contract file, or to an expression holding + it, found to a fixed point across modules: a name bound to one imported from elsewhere is + one too (change review r2, plant N1).""" + exported: dict[str, frozenset[str]] = {} + docs = {path: _docstrings(tree) for path, tree in trees.items()} + owns = {path: _own_nodes(tree) for path, tree in trees.items()} + while True: + found: dict[str, frozenset[str]] = {} + for path, tree in trees.items(): + module_imports = imports[path] + names = _marker_names(tree, module_imports, exported) + + def seed( + sub: ast.AST, + path: Path = path, + names: frozenset[str] = names, + module_imports: _Module = module_imports, + ) -> bool: + return _names_the_file(sub, docs[path], names, module_imports, exported) + + # A name bound to a marker by an import (from .contract import CONTRACT_FILE as + # CONTRACT_JSON) is one too, as W7 says (change review r3, B-M1). + found[module_imports.name] = frozenset( + _bound(owns[path], set(), seed) | (names - MARKER_NAMES) + ) + if found == exported: + return exported + exported = found + + +def scan_contract_writers(root: Path) -> dict[tuple[str, str, str], int]: + trees = {path: ast.parse(path.read_text(encoding="utf-8")) for path in sorted(root.rglob("*.py"))} + package = frozenset(_module_name(path, root) for path in trees) + imports = {path: _context(tree, path, root, package) for path, tree in trees.items()} + helpers = _writer_helpers(trees, imports) + exported = _exported_markers(trees, imports) + hits: dict[tuple[str, str, str], int] = {} + for path, tree in trees.items(): + docs = _docstrings(tree) + module = ".".join(path.relative_to(root.parent.parent).with_suffix("").parts) + module_imports = imports[path] + marker_names = _marker_names(tree, module_imports, exported) + + def names_it( + sub: ast.AST, + docs: set[int] = docs, + marker_names: frozenset[str] = marker_names, + module_imports: _Module = module_imports, + ) -> bool: + return _names_the_file(sub, docs, marker_names, module_imports, exported) + + # The constants each class of this module binds to the file in its body, read as + # Names.CONTRACT_JSON (change review r3). + constants = { + node.name: frozenset(_bound(_own_nodes(node), set(), names_it)) + for node in ast.walk(tree) + if isinstance(node, ast.ClassDef) + } + + def is_marker( + sub: ast.AST, + attrs: frozenset[str] = frozenset(), + names_it: Any = names_it, + constants: dict[str, frozenset[str]] = constants, + ) -> bool: + if isinstance(sub, ast.Attribute) and isinstance(sub.value, ast.Name): + if sub.value.id in _SELF and sub.attr in attrs: + return True + if sub.attr in constants.get(sub.value.id, frozenset()): + return True + return names_it(sub) + + def record( + qualified: str, + own: list[ast.AST], + names: set[str], + attrs: frozenset[str], + module: str = module, + module_imports: _Module = module_imports, + is_marker: Any = is_marker, + ) -> None: + mentions = any( + is_marker(sub, attrs) or (isinstance(sub, ast.Name) and sub.id in names) + for sub in own + ) + if not mentions: + return + verbs: Counter[str] = Counter() + for sub in own: + found = _verb(sub, helpers, module_imports) if isinstance(sub, ast.Call) else None + if found: + verbs[found[0]] += 1 + for verb, count in verbs.items(): + hits[(module, qualified, verb)] = count + + def visit( + node: ast.AST, + marked: set[str], + prefix: str, + attrs: frozenset[str], + is_marker: Any = is_marker, + record: Any = record, + ) -> None: + for child in ast.iter_child_nodes(node): + if isinstance(child, _DEFS): + own = _own_nodes(child) + names = _bound(own, marked, lambda sub, attrs=attrs: is_marker(sub, attrs)) + qualified = prefix + child.name + record(qualified, own, names, attrs) + visit(child, names, qualified + ".", attrs) + elif isinstance(child, ast.ClassDef): + held = _self_attrs(child, marked, is_marker) + own = _own_nodes(child) + names = _bound(own, marked, lambda sub, held=held: is_marker(sub, held)) + record(prefix + child.name + ".
", own, names, held) + visit(child, marked, prefix + child.name + ".", held) + else: + visit(child, marked, prefix, attrs) + + own = _own_nodes(tree) + marked = _bound(own, set(), is_marker) + record("