From b8f17e8e7098d54e5c964c37ce4d97d0d94df386 Mon Sep 17 00:00:00 2001 From: kokokoXUY <13682395396@163.com> Date: Tue, 29 Sep 2026 22:12:14 +0800 Subject: [PATCH] test(quota): pin the prose no-replan contract at the public CLI boundary The existing test calls `typed_progress_repeat_trigger` directly, but the path a user exercises is `quota should-run` through `autonomous_replan_obligation_from_runs`, which projects the "all" history window. A prose compatibility matcher restored only at that boundary keeps the helper test green while a healthy goal receives `autonomous_replan_required` again (#4336). Add an isolated active goal with an open agent todo and four consecutive successful free-text runs whose only content is "installed" / "uninstalled" / "installation completed", then assert through the real public CLI that the packet carries no replan obligation and no `typed_progress_repeat` marker. The typed repeat control runs the same fixture, so the guard is not vacuous. Verified by mutation: restoring a prose fallback for `operation == "all"` only makes these three cases fail while the helper-level test still passes. Signed-off-by: kokokoXUY <13682395396@163.com> --- .../test_progress_observation.py | 193 ++++++++++++++++++ 1 file changed, 193 insertions(+) diff --git a/tests/control_plane/test_progress_observation.py b/tests/control_plane/test_progress_observation.py index 5b6eb19a2e..e4fc320ba3 100644 --- a/tests/control_plane/test_progress_observation.py +++ b/tests/control_plane/test_progress_observation.py @@ -1,7 +1,15 @@ from __future__ import annotations +import json +from pathlib import Path + import pytest +from loopx.control_plane.testing.canary_harness import ( + run_json_cli_result, + write_fixture_registry, +) + from loopx.control_plane.work_items.progress_observation import ( build_replan_action_packet, build_replan_context, @@ -466,3 +474,188 @@ def test_host_projects_evidence_context_and_minimal_action_packet() -> None: "blocked", "no_followup", ] + + +def test_prose_progress_summaries_never_trigger_replan() -> None: + """#4336: an `installed` progress summary was read as `stalled`. + + A 0.4.3 install reported `installed` in a progress summary, the prose matcher + read that as a stall, and a healthy goal received `autonomous_replan_required`. + Typed progress observations replaced the matcher after #3161. This pins the + contract that only a typed observation can trigger the repeat rule, so a + compatibility matcher cannot bring the prose path back silently. + """ + + prose_runs = [ + { + "generated_at": f"2026-08-13T01:0{index}:00Z", + "agent_id": AGENT_ID, + "summary": summary, + "note": summary, + "status": "ok", + } + for index, summary in enumerate( + ( + "installed", + "uninstalled", + "installation completed", + "not installed; stalled on the same route again", + ) + ) + ] + + assert typed_progress_repeat_trigger(prose_runs, agent_id=AGENT_ID) is None + + # An unreadable historical row is not silently upgraded into typed truth + # either: prose inside a semantic identifier keeps the run out of the window. + unreadable = [ + dict(run, progress_observation=dict(observation, surface_id="look at the same route again")) + for run in ( + _run("2026-08-13T01:01:00Z", _observation()), + _run("2026-08-13T01:00:00Z", _observation()), + ) + for observation in (run["progress_observation"],) + ] + + assert typed_progress_repeat_trigger(unreadable, agent_id=AGENT_ID) is None + + # The typed path still triggers, so the guard above is not vacuous. + typed_runs = [ + _run("2026-08-13T01:01:00Z", _observation()), + _run("2026-08-13T01:00:00Z", _observation()), + ] + trigger = typed_progress_repeat_trigger(typed_runs, agent_id=AGENT_ID) + + assert trigger is not None + assert trigger["kind"] == "typed_progress_repeat" + +GOAL_ID = "prose-progress-cli-fixture" +TODO_ID = "todo_prose_progress_slice" + + +def _cli_fixture(root: Path) -> tuple[Path, Path]: + """Return one isolated active Goal whose only run history is free text.""" + + runtime = root / "runtime" + registry = root / "registry.json" + state_file = root / "state.md" + state_file.write_text( + "---\nstatus: active\n---\n" + "# Prose progress fixture\n\n" + "## Objective\n\n" + "Keep a healthy installation moving.\n\n" + "## Agent Todo\n\n" + "- [ ] [P1] Continue the bounded slice.\n" + f" \n", + encoding="utf-8", + ) + write_fixture_registry( + project=root, + runtime_root=runtime, + registry_path=registry, + goal_id=GOAL_ID, + domain="prose-progress", + adapter_kind="generic_project_goal_v0", + state_file=str(state_file), + registered_agents=[AGENT_ID], + quota_allowed_slots=None, + ) + return registry, runtime + + +def _write_index_jsonl(runtime: Path, newest_first_runs: list[dict[str, object]]) -> None: + index = runtime / "goals" / GOAL_ID / "runs" / "index.jsonl" + index.parent.mkdir(parents=True, exist_ok=True) + index.write_text( + "".join(json.dumps(run, sort_keys=True) + "\n" for run in newest_first_runs), + encoding="utf-8", + ) + + +def _quota_should_run(registry: Path, runtime: Path, scan_path: Path) -> dict[str, object]: + code, packet = run_json_cli_result( + "quota", + "should-run", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--scan-path", + str(scan_path), + registry_path=registry, + runtime_root=runtime, + ) + assert code == 0, packet + return packet + + +def _successful_prose_run(generated_at: str, summary: str) -> dict[str, object]: + return { + "generated_at": generated_at, + "goal_id": GOAL_ID, + "agent_id": AGENT_ID, + "classification": "bounded_replan_progress", + "turn_instance_id": f"turn-{generated_at}", + "status": "ok", + "summary": summary, + "note": summary, + } + + +@pytest.mark.parametrize("summary", ["installed", "uninstalled", "installation completed"]) +def test_public_quota_never_replans_from_prose_run_history(tmp_path: Path, summary: str) -> None: + """#4336 at the public boundary: the real `quota should-run` stays quiet. + + The helper-level test above pins the codec. This one pins the consumer a user + actually runs, so a prose compatibility matcher reintroduced anywhere below + the public CLI fails here even while the helper test stays green. + """ + + registry, runtime = _cli_fixture(tmp_path) + _write_index_jsonl( + runtime, + [ + _successful_prose_run("2026-09-01T00:03:00+00:00", summary), + _successful_prose_run("2026-09-01T00:02:00+00:00", summary), + _successful_prose_run("2026-09-01T00:01:00+00:00", "uninstalled"), + _successful_prose_run("2026-09-01T00:00:00+00:00", "installation completed"), + ], + ) + + packet = _quota_should_run(registry, runtime, tmp_path) + + assert packet["decision"] != "autonomous_replan_required" + assert packet.get("replan_action_packet") is None + assert packet["requires_user_action"] is False + serialized = json.dumps(packet) + assert "autonomous_replan" not in serialized + assert "typed_progress_repeat" not in serialized + + +def test_public_quota_still_replans_from_typed_run_history(tmp_path: Path) -> None: + """The typed repeat rule keeps reaching the same public consumer.""" + + registry, runtime = _cli_fixture(tmp_path) + _write_index_jsonl( + runtime, + [ + { + **_successful_prose_run("2026-09-01T00:01:00+00:00", "installed"), + "progress_observation": normalize_progress_observation(_observation()), + }, + { + **_successful_prose_run("2026-09-01T00:00:00+00:00", "installed"), + "progress_observation": normalize_progress_observation(_observation()), + }, + ], + ) + + packet = _quota_should_run(registry, runtime, tmp_path) + obligation = packet["replan_action_packet"] + + assert packet["decision"] == "autonomous_replan_required" + assert isinstance(obligation, dict) + assert obligation["decision"] == "replan_required" + assert str(obligation["obligation_id"]).startswith("replan-") + assert obligation["required_outcome"]