fix(harness): reseal the scenarios that load and name the ones that could not be applied (TH-8929)

src/fi/alk/harness/bundle_author_v2.py

@@ -23,7 +23,7 @@
 import tomllib
 from dataclasses import dataclass, replace
 from pathlib import Path
-from typing import Any, Sequence
+from typing import Any, Iterable, Sequence
 from urllib.parse import urlsplit
 
 import yaml
@@ -3226,7 +3226,24 @@ def exists_from(root: Path, path: str) -> bool:
     return replace(plan, processes=tuple(processes))
 
 
-def _copy_scenarios(authoring: Path, staging: Path, *, count: int) -> None:
+def _folder_key(folder: Path) -> str:
+    try:
+        body = json.loads((folder / "scenario.json").read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return folder.name
+    return str(body.get("scenario_key") or body.get("name") or folder.name) if isinstance(body, dict) else folder.name
+
+
+def _copy_scenarios(
+    authoring: Path,
+    staging: Path,
+    *,
+    count: int,
+    skip: frozenset[str] = frozenset(),
+    sealed: Path | None = None,
+) -> None:
+    """Copy the authoring scenarios into `staging`. A folder whose key is in `skip` keeps the copy
+    already sealed in `sealed` (its last good version), or is left out when it was never sealed."""
     source = authoring / "scenarios"
     if not source.is_dir():
         raise BundleAuthorError(f"scenario_artifacts_missing: {source}")
@@ -3237,7 +3254,19 @@ def _copy_scenarios(authoring: Path, staging: Path, *, count: int) -> None:
         raise BundleAuthorError(
             f"scenario_artifacts_insufficient: requested {count}, found {len(folders)}"
         )
+    held = (
+        {_folder_key(path): path for path in (sealed / "scenarios").iterdir() if path.is_dir()}
+        if skip and sealed is not None and (sealed / "scenarios").is_dir()
+        else {}
+    )
+    # A held copy never takes a folder name a case being written now uses; it is left out instead.
+    written = {path.name for path in folders[:count] if _folder_key(path) not in skip}
     for source_folder in folders[:count]:
+        key = _folder_key(source_folder)
+        if key in skip:
+            if key in held and held[key].name not in written:
+                shutil.copytree(held[key], target / held[key].name)
+            continue
         shutil.copytree(source_folder, target / source_folder.name)
     for folder in sorted(path for path in target.iterdir() if path.is_dir()):
         document = folder / "scenario.json"
@@ -3887,16 +3916,38 @@ def author_bundle_v2(
         raise
 
 
-def reseal_scenarios(authoring: str | Path) -> str | None:
+def scenario_seal_problems(authoring: str | Path) -> dict[str, str]:
+    """Each scenario folder a run could not load, by key, with the first line of why. Loaded one
+    folder at a time, the way a run loads them, so one broken case does not hide the rest."""
+    from .scenario_source import _declared_tool_names, _deterministic_names, _load_one
+
+    root = Path(authoring)
+    folders = sorted(path for path in (root / "scenarios").iterdir() if path.is_dir()) if (root / "scenarios").is_dir() else []
+    if not folders:
+        return {}
+    settled_in_code = _deterministic_names(root)
+    declared_tools = _declared_tool_names(root)
+    problems: dict[str, str] = {}
+    for folder in folders:
+        try:
+            _load_one(folder, settled_in_code=settled_in_code, declared_tools=declared_tools, defer_execution=False)
+        except Exception as exc:  # noqa: BLE001 - every reason a run would refuse this folder
+            problems[_folder_key(folder)] = (str(exc).strip().splitlines() or [type(exc).__name__])[0][:300]
+    return problems
+
+
+def reseal_scenarios(authoring: str | Path, *, skip: Iterable[str] = ()) -> str | None:
     """Carry an edited scenario set into the sealed bundle and seal it again.
 
     Runs replay the bundle, never the authoring folders, so a scenario added or rewritten after
     the build would otherwise be unknown to every Run. Only the scenario-owned files move; the
     runtime, seed and processes stay exactly what was built and validated. A generic harness's
     certificate binds the bundle digest, so it is reissued for the new set with the edit recorded
     as a limitation: those scenarios were proved against the authoring world, not the runtime.
-    Returns the new digest, or None when there is no sealed bundle to update.
+    Returns the new digest, or None when there is no sealed bundle to update. A scenario whose key
+    is in `skip` keeps its last sealed version (or stays out if it was never sealed).
     """
+    skip = frozenset(skip)
     authoring_root = Path(authoring)
     bundle = authoring_root / "environment-bundle"
     if not (bundle / BUNDLE_V2_MANIFEST).is_file():
@@ -3909,7 +3960,7 @@ def reseal_scenarios(authoring: str | Path) -> str | None:
         shutil.copytree(bundle, staging)
         (staging / BUNDLE_V2_MANIFEST).unlink()
         shutil.rmtree(staging / "scenarios", ignore_errors=True)
-        _copy_scenarios(authoring_root, staging, count=count)
+        _copy_scenarios(authoring_root, staging, count=count, skip=skip, sealed=bundle)
         if (authoring_root / CATALOGUE).is_file():
             shutil.copy2(authoring_root / CATALOGUE, staging / CATALOGUE)
         compiled, processes, metadata = _carry_contract(authoring_root, staging, manifest)

src/fi/alk/harness/hosted_chat_entrypoint.py

@@ -1054,10 +1054,20 @@ async def _save_changes(self, *, force: bool = False) -> dict[str, Any]:
         fingerprint = self._environment_fingerprint()
         if fingerprint == self._published_fingerprint() and not force:
             return {"applied": True, "unchanged": True, "message": "nothing is waiting to be saved"}
-        from .bundle_author_v2 import reseal_scenarios
+        from .bundle_author_v2 import reseal_scenarios, scenario_seal_problems
 
+        # A scenario a run could not load is left out of the seal (it keeps its last sealed
+        # version) and named to the platform, so one broken case never blocks the rest.
         try:
-            await asyncio.to_thread(reseal_scenarios, self.workspace)
+            problems = await asyncio.to_thread(scenario_seal_problems, self.workspace)
+        except Exception:  # noqa: BLE001 - an unreadable tree is the reseal's to report
+            problems = {}
+        failed = [{"scenario_key": key, "reason": reason} for key, reason in sorted(problems.items())]
+        try:
+            if problems:
+                await asyncio.to_thread(reseal_scenarios, self.workspace, skip=set(problems))
+            else:
+                await asyncio.to_thread(reseal_scenarios, self.workspace)
         except Exception as exc:  # noqa: BLE001 - reported to the model, never swallowed
             traceback.print_exc()
             return {
@@ -1070,7 +1080,12 @@ async def _save_changes(self, *, force: bool = False) -> dict[str, Any]:
             try:
                 async with self._checkpoint_lock:
                     stored = await self.client.checkpoint(self.workspace)
-                    result = await self.client.publish(str(stored.get("digest") or ""))
+                    digest = str(stored.get("digest") or "")
+                    result = (
+                        await self.client.publish(digest, failed_scenarios=failed)
+                        if failed
+                        else await self.client.publish(digest)
+                    )
             except ConversationTransportError as exc:
                 return {"applied": False, "code": "publish_unreachable", "message": str(exc)}
             if result.get("code") != "conversation_checkpoint_not_latest":
@@ -1083,6 +1098,8 @@ async def _save_changes(self, *, force: bool = False) -> dict[str, Any]:
                 "loaded, so these changes were not saved. The chat reloads the current "
                 "environment after this reply; the change has to be made again on it."
             )
+        if result.get("applied") and failed:
+            result["failed"] = failed
         if result.get("applied"):
             await self._record_touched(result)
             if result.get("revision"):
@@ -1100,10 +1117,16 @@ async def _publish_if_changed(self) -> tuple[str, bool]:
             return "", False
         result = await self._save_changes()
         if not result.get("applied"):
+            if result.get("code") == "reseal_failed":
+                # Not a fault of one case: the change stays in this chat's working copy as a draft.
+                return (
+                    "These changes are kept as an unsaved draft and were NOT saved to the environment: "
+                    f"{result.get('message') or 'they could not be sealed for runs'}."
+                ), True
             return (
                 "These changes were NOT saved to the environment: "
                 f"{result.get('message') or result.get('code') or 'unknown reason'}."
-            ), result.get("code") == "reseal_failed"
+            ), False
 
         def named(rows: list[dict[str, Any]]) -> str:
             return ", ".join(f"#{one.get('number')} {one.get('name')}" for one in rows)
@@ -1113,10 +1136,19 @@ def named(rows: list[dict[str, Any]]) -> str:
             for label, key in (("added", "added"), ("removed", "removed"), ("changed", "changed"))
             if result.get(key)
         ]
+        failed = result.get("failed") or []
+        except_line = (
+            f" Saved, except {len(failed)} scenarios that could not be applied: "
+            + "; ".join(f"{one['scenario_key']}: {one['reason']}" for one in failed)
+            + "."
+            if failed
+            else ""
+        )
         return (
             "Saved to the environment: "
             + ("; ".join(parts) or "no scenario changed")
             + f". It now has {result.get('scenario_count')} scenarios."
+            + except_line
         ), False
 
     async def _selection_note(self, payload: dict[str, Any]) -> str:
@@ -1673,22 +1705,11 @@ def receive(event: Event) -> None:
             )
             return
         final_text = "".join(emitted_text).strip()
-        saved, unsealed = await self._publish_if_changed()
+        # Changes no run could replay (a contract or seed fault, not one case) are kept: the turn
+        # completes and its checkpoint is promoted as the chat's draft, unsaved to the environment.
+        saved, _unsealed = await self._publish_if_changed()
         if saved:
             final_text = f"{final_text}\n\n{saved}".strip()
-        if unsealed:
-            # Changes no run could replay are not promoted with this turn.
-            await self._fail_turn(
-                journal,
-                message_id,
-                assistant_message_id,
-                invocation_id,
-                final_text,
-                "reseal_failed",
-            )
-            if self._restart_after_turn:
-                self.stopping.set()
-            return
         if self._last_health and not self._last_health.startswith("Healthy") and not self._control_only():
             problems = [line for line in self._last_health.splitlines() if line.strip()]
             final_text = (

src/fi/alk/harness/hosted_conversation.py

@@ -136,6 +136,9 @@ def load_conversation_capabilities(
     return capabilities
 
 
+FAILED_SCENARIOS_LIMIT = 200
+
+
 class ConversationClient:
     def __init__(
         self,
@@ -259,8 +262,14 @@ async def run_status(self) -> dict[str, Any]:
     async def scenarios(self, query: dict[str, Any]) -> dict[str, Any]:
         return await self._platform("scenarios", query)
 
-    async def publish(self, digest: str) -> dict[str, Any]:
-        return await self._platform("publish", {"digest": digest})
+    async def publish(self, digest: str, *, failed_scenarios: list[dict[str, str]] | None = None) -> dict[str, Any]:
+        """Publish a checkpoint. `failed_scenarios` ({scenario_key, reason}) names cases left out of the
+        seal because a run could not load them; sent only when there are any, at most the platform's
+        200 so a larger set never refuses the whole save."""
+        body: dict[str, Any] = {"digest": digest}
+        if failed_scenarios:
+            body["failed_scenarios"] = list(failed_scenarios)[:FAILED_SCENARIOS_LIMIT]
+        return await self._platform("publish", body)
 
     async def runs(self, run_id: str = "", limit: int = 10) -> dict[str, Any]:
         return await self._platform(

tests/harness/test_builder_chat.py

@@ -108,6 +108,8 @@ def _runtime(tmp_path: Path, client: BuilderClient, monkeypatch) -> HostedChatRu
     monkeypatch.setenv("ALK_GOOGLE_ADK_EXPLICITLY_APPROVED_BY_USER", "1")
     (tmp_path / "scenarios" / "one").mkdir(parents=True)
     (tmp_path / "scenarios" / "one" / "scenario.json").write_text("{}")
+    # These stub folders are not real scenarios; a test about one a run cannot load says so itself.
+    monkeypatch.setattr("fi.alk.harness.bundle_author_v2.scenario_seal_problems", lambda root: {})
     policy = tmp_path / "policy.yaml"
     policy.write_text("schema_version: 1\nharness_chat:\n  enabled: true\n", encoding="utf-8")
     runtime = HostedChatRuntime(
@@ -189,9 +191,7 @@ def test_a_rejected_publish_is_never_reported_as_done(tmp_path, monkeypatch):
     assert not (tmp_path / ".futureagi-authoring-basis.json").exists()
 
 
-def test_a_failed_reseal_ends_the_turn_failed_with_nothing_to_land(
-    tmp_path, monkeypatch
-):
+def test_a_failed_reseal_keeps_the_change_as_a_draft(tmp_path, monkeypatch):
     client = BuilderClient()
     runtime = _runtime(tmp_path, client, monkeypatch)
     runtime.conversation = EditingConversation(tmp_path, writes=True)
@@ -206,12 +206,13 @@ def fails(root):
 
     assert client.published == []
     reply = next(e for e in client.events if e["kind"] == "assistant_message")
-    assert "These changes were NOT saved" in reply["payload"]["text"]
-    assert reply["payload"]["error"] == "reseal_failed"
+    assert "kept as an unsaved draft and were NOT saved" in reply["payload"]["text"]
+    assert "seed import failed" in reply["payload"]["text"]
     kinds = [event["kind"] for event in client.events]
-    assert "checkpoint_committed" not in kinds
+    # The turn completes and its checkpoint is promoted as the chat's draft.
+    assert "checkpoint_committed" in kinds
     assert client.events[-1]["kind"] == "turn_completed"
-    assert client.events[-1]["payload"]["outcome"] == "failed"
+    assert client.events[-1]["payload"]["outcome"] != "failed"
 
 
 def test_read_only_tools_skip_the_workspace_checkpoint(tmp_path, monkeypatch):
@@ -2163,3 +2164,97 @@ def test_exact_claim_gets_no_correction():
         )
         is None
     )
+
+
+def test_one_broken_scenario_does_not_block_the_save(tmp_path, monkeypatch):
+    class FailedAwareClient(BuilderClient):
+        def __init__(self):
+            super().__init__()
+            self.failed: list = []
+
+        async def publish(self, digest, failed_scenarios=None):
+            self.failed.append(failed_scenarios)
+            return await super().publish(digest)
+
+    client = FailedAwareClient()
+    runtime = _runtime(tmp_path, client, monkeypatch)
+    runtime.conversation = EditingConversation(tmp_path, writes=True)
+    runtime._published_fingerprint()
+    sealed: list = []
+    monkeypatch.setattr(
+        "fi.alk.harness.bundle_author_v2.scenario_seal_problems",
+        lambda root: {"broken_case": "checks/refunded.py would not compile"},
+    )
+    monkeypatch.setattr(
+        "fi.alk.harness.bundle_author_v2.reseal_scenarios", lambda root, skip=(): sealed.append(set(skip))
+    )
+
+    _turn(runtime, {"content": "edit three cases"})
+
+    assert sealed == [{"broken_case"}]
+    assert client.failed == [[{"scenario_key": "broken_case", "reason": "checks/refunded.py would not compile"}]]
+    text = _reply(client)
+    assert "Saved to the environment: added #41 late_driver" in text
+    assert "Saved, except 1 scenarios that could not be applied: broken_case: checks/refunded.py would not compile." in text
+
+
+def test_a_folder_a_run_cannot_load_keeps_its_sealed_version(tmp_path):
+    from fi.alk.harness.bundle_author_v2 import _copy_scenarios, scenario_seal_problems
+
+    authoring = tmp_path / "authoring"
+    for name, body in (("good", '{"scenario_key": "good"}'), ("broken", "{not json")):
+        (authoring / "scenarios" / name).mkdir(parents=True)
+        (authoring / "scenarios" / name / "scenario.json").write_text(body, encoding="utf-8")
+    sealed = tmp_path / "bundle"
+    (sealed / "scenarios" / "broken").mkdir(parents=True)
+    (sealed / "scenarios" / "broken" / "scenario.json").write_text('{"scenario_key": "broken", "v": 1}', encoding="utf-8")
+
+    assert "broken" in scenario_seal_problems(authoring)
+    staging = tmp_path / "staging"
+    staging.mkdir()
+    _copy_scenarios(authoring, staging, count=2, skip=frozenset({"broken"}), sealed=sealed)
+
+    assert json.loads((staging / "scenarios" / "broken" / "scenario.json").read_text())["v"] == 1
+    assert (staging / "scenarios" / "good").is_dir()
+
+
+def test_a_held_copy_never_takes_the_folder_of_a_case_written_now(tmp_path):
+    from fi.alk.harness.bundle_author_v2 import _copy_scenarios
+
+    authoring = tmp_path / "authoring"
+    for name, body in (("a_broken", "{not json"), ("late", '{"scenario_key": "late"}')):
+        (authoring / "scenarios" / name).mkdir(parents=True)
+        (authoring / "scenarios" / name / "scenario.json").write_text(body, encoding="utf-8")
+    sealed = tmp_path / "bundle"
+    (sealed / "scenarios" / "late").mkdir(parents=True)
+    (sealed / "scenarios" / "late" / "scenario.json").write_text('{"scenario_key": "a_broken"}', encoding="utf-8")
+    staging = tmp_path / "staging"
+    staging.mkdir()
+
+    _copy_scenarios(authoring, staging, count=2, skip=frozenset({"a_broken"}), sealed=sealed)
+
+    assert json.loads((staging / "scenarios" / "late" / "scenario.json").read_text())["scenario_key"] == "late"
+    assert sorted(path.name for path in (staging / "scenarios").iterdir()) == ["late"]
+
+
+def test_publish_sends_at_most_the_platform_limit_of_failed_scenarios():
+    from types import SimpleNamespace
+
+    body = _capabilities().model_dump(mode="json")
+    body["endpoints"]["publish"] = body["endpoints"]["run_status"].replace("run-status", "publish")
+    sent = []
+
+    class Transport:
+        def request(self, method, url, **kwargs):
+            sent.append(kwargs.get("json_body"))
+            return SimpleNamespace(status_code=200, body={"applied": True})
+
+    client = ConversationClient(ConversationCapabilities.model_validate(body), transport=Transport())
+    failed = [{"scenario_key": f"case-{one}", "reason": "bad check"} for one in range(250)]
+
+    asyncio.run(client.publish("sha256:abc", failed_scenarios=failed))
+    asyncio.run(client.publish("sha256:abc"))
+
+    assert len(sent[0]["failed_scenarios"]) == 200
+    assert sent[0]["failed_scenarios"][0] == {"scenario_key": "case-0", "reason": "bad check"}
+    assert "failed_scenarios" not in sent[1]