From c12abcffe357d6f1c3b8fa9aedefa0494d411131 Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Sun, 15 Nov 2026 01:10:32 +0530 Subject: [PATCH 1/3] fix(harness): a scenario a run cannot load is left out of the seal and named on publish, and a failed reseal keeps the change as a draft --- src/fi/alk/harness/bundle_author_v2.py | 59 +++++++++++++++-- src/fi/alk/harness/hosted_chat_entrypoint.py | 57 +++++++++++------ src/fi/alk/harness/hosted_conversation.py | 9 ++- tests/harness/test_builder_chat.py | 67 ++++++++++++++++++-- 4 files changed, 160 insertions(+), 32 deletions(-) diff --git a/src/fi/alk/harness/bundle_author_v2.py b/src/fi/alk/harness/bundle_author_v2.py index 0ee5f0ada..ef0b8936d 100644 --- a/src/fi/alk/harness/bundle_author_v2.py +++ b/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,17 @@ 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 {} + ) for source_folder in folders[:count]: + key = _folder_key(source_folder) + if key in skip: + if key in held: + 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,7 +3914,27 @@ 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 @@ -3895,8 +3942,10 @@ def reseal_scenarios(authoring: str | Path) -> str | None: 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 +3958,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) diff --git a/src/fi/alk/harness/hosted_chat_entrypoint.py b/src/fi/alk/harness/hosted_chat_entrypoint.py index b873cf87b..1a595b412 100644 --- a/src/fi/alk/harness/hosted_chat_entrypoint.py +++ b/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 = ( diff --git a/src/fi/alk/harness/hosted_conversation.py b/src/fi/alk/harness/hosted_conversation.py index a03bfda5c..6a02b6dea 100644 --- a/src/fi/alk/harness/hosted_conversation.py +++ b/src/fi/alk/harness/hosted_conversation.py @@ -259,8 +259,13 @@ 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.""" + body: dict[str, Any] = {"digest": digest} + if failed_scenarios: + body["failed_scenarios"] = list(failed_scenarios) + return await self._platform("publish", body) async def runs(self, run_id: str = "", limit: int = 10) -> dict[str, Any]: return await self._platform( diff --git a/tests/harness/test_builder_chat.py b/tests/harness/test_builder_chat.py index 0ebe419b3..a2e54028f 100644 --- a/tests/harness/test_builder_chat.py +++ b/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): @@ -2145,3 +2146,55 @@ 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() From 1f150a0cefb9a9cb5f992ff1bf69edfb66970548 Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Sun, 15 Nov 2026 04:43:26 +0530 Subject: [PATCH 2/3] fix(harness): send at most the platform's 200 failed scenarios with a publish so a larger set never refuses the save --- src/fi/alk/harness/hosted_conversation.py | 8 ++++++-- tests/harness/test_builder_chat.py | 23 +++++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/src/fi/alk/harness/hosted_conversation.py b/src/fi/alk/harness/hosted_conversation.py index 6a02b6dea..18fd1362c 100644 --- a/src/fi/alk/harness/hosted_conversation.py +++ b/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, @@ -261,10 +264,11 @@ async def scenarios(self, query: dict[str, Any]) -> dict[str, Any]: 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.""" + 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) + 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]: diff --git a/tests/harness/test_builder_chat.py b/tests/harness/test_builder_chat.py index a2e54028f..f3ec3336d 100644 --- a/tests/harness/test_builder_chat.py +++ b/tests/harness/test_builder_chat.py @@ -2198,3 +2198,26 @@ def test_a_folder_a_run_cannot_load_keeps_its_sealed_version(tmp_path): assert json.loads((staging / "scenarios" / "broken" / "scenario.json").read_text())["v"] == 1 assert (staging / "scenarios" / "good").is_dir() + + +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] From 0583072aaae2274ff5ba48b4833c79ded70a7e3e Mon Sep 17 00:00:00 2001 From: KarthikAvinashFI Date: Sun, 15 Nov 2026 07:00:37 +0530 Subject: [PATCH 3/3] fix(harness): a held scenario's last sealed copy never takes the folder of a case written in the same save --- src/fi/alk/harness/bundle_author_v2.py | 4 +++- tests/harness/test_builder_chat.py | 19 +++++++++++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/src/fi/alk/harness/bundle_author_v2.py b/src/fi/alk/harness/bundle_author_v2.py index ef0b8936d..f48460903 100644 --- a/src/fi/alk/harness/bundle_author_v2.py +++ b/src/fi/alk/harness/bundle_author_v2.py @@ -3259,10 +3259,12 @@ def _copy_scenarios( 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: + 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) diff --git a/tests/harness/test_builder_chat.py b/tests/harness/test_builder_chat.py index f3ec3336d..a692bb82d 100644 --- a/tests/harness/test_builder_chat.py +++ b/tests/harness/test_builder_chat.py @@ -2200,6 +2200,25 @@ def test_a_folder_a_run_cannot_load_keeps_its_sealed_version(tmp_path): 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