feat(harness): let a build's first run reuse the agent reading (TH-8952)

.github/workflows/sdk-smoke.yml

@@ -113,6 +113,7 @@ jobs:
           tests/harness/test_scan_ai.py
           tests/harness/test_agent_scan.py
           tests/harness/test_authoring_entrypoint_find_agents.py
+          tests/harness/test_seed_contract.py
           tests/test_bump_deployment_values.py
           tests/test_release_workflow.py
           tests/test_release_pr_step.py

src/fi/alk/harness/authoring_entrypoint.py

@@ -690,7 +690,18 @@ def main(argv: list[str] | None = None, *, validate_runtime: bool = False) -> in
         action="store_true",
         help="Find the agents in the source and leave their list in --output; build nothing",
     )
+    parser.add_argument(
+        "--seed-contract",
+        type=Path,
+        help="A reading's raw contract.json at /work/seed/contract.json; the build reuses it when it matches the job",
+    )
     args = parser.parse_args(argv)
+    if args.seed_contract is not None and (
+        args.certify_only or args.read_agent_only or args.find_agents
+    ):
+        parser.error(
+            "--seed-contract cannot be combined with --certify-only, --read-agent-only or --find-agents"
+        )
     if args.certify_only:
         if args.bundle_output is None:
             parser.error("--certify-only requires --bundle-output")
@@ -729,6 +740,22 @@ def main(argv: list[str] | None = None, *, validate_runtime: bool = False) -> in
         args.source.resolve(), job.source.subfolder, args.output.resolve()
     ):
         return EXIT_REFUSED
+    from .seed_contract import SEED_ACCEPTED
+
+    # Cleared before the seed checks: a mark an earlier command left must not pass for this one's yes.
+    (args.output / SEED_ACCEPTED).unlink(missing_ok=True)
+    if args.seed_contract is not None:
+        if args.await_harness:
+            from .seed_contract import accept_seed
+
+            refused = accept_seed(args.seed_contract, args.output)
+            if refused is not None:
+                return refused
+        else:
+            print(
+                "seed contract: not used; only a build the conversation's harness runs takes one",
+                file=sys.stderr,
+            )
     broker_socket = initialize_hosted_compose(job, args.source.resolve())
     if broker_socket is not None:
         os.environ["ALK_COMPOSE_BROKER_SOCKET"] = str(broker_socket)

src/fi/alk/harness/hosted_chat_entrypoint.py

@@ -7,6 +7,8 @@
 import json
 import os
 import signal
+import sys
+import threading
 import time
 import uuid
 from collections.abc import Awaitable
@@ -57,6 +59,8 @@
 _PUBLISHED = ".futureagi-published.json"
 _LEDGER = ".futureagi-ledger.json"
 _LEDGER_NOTES = 40
+# One install at a time: an opening cancelled mid-install leaves its thread running.
+_SEED_LOCK = threading.Lock()
 # Tools whose effect the person confirms before it happens.
 _NEEDS_CONFIRMATION = frozenset(
     {
@@ -269,6 +273,10 @@ async def close(self) -> None:
 
 
 class HostedChatRuntime:
+    # Whether this process's first opening has decided on the seed, and what it decided.
+    _seed_taken = False
+    _seed_outcome: tuple[str, list[str]] | None = None
+
     def __init__(
         self,
         *,
@@ -341,9 +349,24 @@ async def _open_builder(self, *, resume: bool = True) -> Stage:
         from .understand import load as load_contract
 
         policy_job = guest_booking_policy_job(self.job)
+        # A reading the platform handed this build is used before anything opens.
+        await asyncio.to_thread(self._take_seed)
         contract = load_contract(self.workspace)
         self._opened_with = self._what_exists()
         if not self._opened_with[1]:
+            if contract is not None:
+                try:
+                    # What follows a saved contract is done before the world stage opens.
+                    await asyncio.to_thread(self._after_saved)
+                except Exception as exc:  # noqa: BLE001 - the world stage opens without it; it is tried again after a turn, at finish_build and at the next opening
+                    from .seed_contract import print_trace
+
+                    print(
+                        f"source model not written before the world stage; class: {type(exc).__name__}",
+                        file=sys.stderr,
+                        flush=True,
+                    )
+                    print_trace(exc)
             stage = self._open_for_building(contract)
         else:
             try:
@@ -1263,6 +1286,87 @@ def _after_saved(self, *, world_changed: bool = False) -> None:
         if has_world and (world_changed or not (self.workspace / "environment-bundle" / "manifest.json").is_file()):
             after_world(self.source if self.source.is_dir() else None, self.workspace, load_contract(self.workspace))
 
+    def _take_seed(self) -> None:
+        """Use the reading the platform handed this build, at the first opening only, and leave
+        what was decided for the platform."""
+        from .seed_contract import SEED_RESULT, write_result
+
+        with _SEED_LOCK:
+            if self._seed_taken:
+                # Decided at this process's first opening; a record that couldn't be written is
+                # tried again, but nothing is decided twice.
+                if self._seed_outcome is not None and not os.path.lexists(self.workspace / SEED_RESULT):
+                    write_result(self.workspace, *self._seed_outcome)
+                return
+            self._seed_taken = True
+            self._seed_outcome = self._decide_seed()
+            if self._seed_outcome is not None:
+                write_result(self.workspace, *self._seed_outcome)
+
+    def _decide_seed(self) -> tuple[str, list[str]] | None:
+        """What the first opening does with the seed, or None when there is nothing to say."""
+        from .seed_contract import (
+            SEED_ACCEPTED,
+            SEED_INSTALLED,
+            SEED_RESULT,
+            holds_seed,
+            print_trace,
+            seed_beside,
+        )
+
+        seed = seed_beside(self.workspace)
+        # The platform placed no seed: nothing to decide, and nothing to say.
+        if not os.path.lexists(seed):
+            return None
+        # Decided by an earlier process that could say so: once the file exists, nothing installs.
+        if os.path.lexists(self.workspace / SEED_RESULT):
+            return None
+        outcome: tuple[str, list[str]]
+        contract = self.workspace / "contract.json"
+        if contract.exists() and ((self.workspace / SEED_INSTALLED).is_file() or holds_seed(contract, seed)):
+            # Installed by an earlier process that stopped before it could say so. The record
+            # outlasts a contract edit; before it is written no stage has opened, so the bytes
+            # are still the seed's, which no reading makes, as only the paid one is recorded.
+            outcome = ("used", [])
+        elif (
+            contract.exists()
+            or (self.workspace / FAILURE_FILE).exists()
+            or self._what_exists()[1]
+            or self._build_is_over()
+        ):
+            # Not a fresh seeded build: the build command refused, the workspace was restored,
+            # or a reading of the builder's own has saved.
+            return None
+        elif not (self.workspace / SEED_ACCEPTED).is_file():
+            # Opened before the build command accepted it, so this build reads the agent itself.
+            return "not_accepted", []
+        else:
+            outcome = self._use_seed(seed)
+        try:
+            # Consumed whatever the outcome, so nothing but this opening ever acts on it.
+            (self.workspace / SEED_ACCEPTED).unlink(missing_ok=True)
+        except OSError as exc:
+            print_trace(exc)
+        return outcome
+
+    def _use_seed(self, seed: Path) -> tuple[str, list[str]]:
+        from pydantic import ValidationError
+
+        from .job import HarnessJob
+        from .seed_contract import print_trace, use_seed
+
+        try:
+            job = HarnessJob.model_validate(self.job)
+        except ValidationError:
+            # Only the class: its text repeats values from the job.
+            print("seed contract: error; class: ValidationError", file=sys.stderr, flush=True)
+            return "error", []
+        except Exception as exc:  # noqa: BLE001 - a job the builder can't read reads the agent itself
+            print(f"seed contract: error; class: {type(exc).__name__}", file=sys.stderr, flush=True)
+            print_trace(exc)
+            return "error", []
+        return use_seed(seed, job, self.source, self.workspace)
+
     async def _handle(self, command: dict[str, Any]) -> None:
         if str(command["kind"]) != "nudge":
             await self._handle_command(command)

src/fi/alk/harness/reading_result.py

@@ -205,6 +205,18 @@ def document(job: HarnessJob, source: Path, output: Path) -> int:
             contract_bytes = max(os.fstat(handle.fileno()).st_size, len(raw))
         if contract_bytes > _CONTRACT_LIMIT_BYTES:
             raise _TooLarge({"contract_bytes": contract_bytes})
+        try:
+            from .seed_contract import replace_if_unchanged, with_record
+
+            # What the reading was made from, so a build can tell whether it may reuse it. Made
+            # from these bytes and written only over them, so a later save is never lost.
+            recorded = with_record(job, raw)
+            if recorded is not None:
+                replace_if_unchanged(output / CONTRACT_FILE, raw, recorded)
+        except Exception:
+            # A paid reading never fails for its record. The mask covers the trace, and
+            # nothing here runs pydantic.
+            logger.warning("reading record not written", exc_info=True)
         contract_text = raw.decode("utf-8")
         framework = detect_framework(
             _framework_folder(job, source, output, contract_text),