[TH-9454] fix: follow-ups from #245's last review of the agent reading

.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'

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

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]:

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"))

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

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,
         )