From 28a18f60cd39832f1262f2576b0c01f85527d2ed Mon Sep 17 00:00:00 2001 From: mika <211269698+mikamikasuki@users.noreply.github.com> Date: Thu, 8 Oct 2026 13:17:18 -0700 Subject: [PATCH] test(periodic-report): align CLI cases with exact refresh contract Signed-off-by: mika <211269698+mikamikasuki@users.noreply.github.com> --- .../test_periodic_report_native_successor.py | 3 +- ...periodic_report_intent_capability_chain.py | 60 ++++++------------- ...t_todo_complete_settlement_capabilities.py | 27 +++------ 3 files changed, 28 insertions(+), 62 deletions(-) diff --git a/tests/capabilities/test_periodic_report_native_successor.py b/tests/capabilities/test_periodic_report_native_successor.py index d4fb487496..c3328274c9 100644 --- a/tests/capabilities/test_periodic_report_native_successor.py +++ b/tests/capabilities/test_periodic_report_native_successor.py @@ -41,7 +41,7 @@ def _native_successor( "--delivery-workspace-path", str(project), "--agent-id", AGENT_ID, "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress", "--no-global-sync", - "--suppress-external-sinks", + "--suppress-external-sinks", "--available-capability", "network", ] closed = _run_cli([ *base, "--classification", "fixture_stage_closed", @@ -141,6 +141,7 @@ def test_native_successor_produces_only_an_enabled_bounded_intent( assert intent["requested_write_scope"] == [] assert intent["payload"]["generation_authorized"] is False assert intent["payload"]["external_delivery_authorized"] is False + assert intent["payload"]["available_capabilities"] == ["network"] stage = intent["payload"]["stage_completion"] assert stage["transition"] == "successor_frontier_settled" assert stage["acceptance"] == "validated" diff --git a/tests/cli_commands/test_periodic_report_intent_capability_chain.py b/tests/cli_commands/test_periodic_report_intent_capability_chain.py index ce8b04f5b9..9ad4ec67da 100644 --- a/tests/cli_commands/test_periodic_report_intent_capability_chain.py +++ b/tests/cli_commands/test_periodic_report_intent_capability_chain.py @@ -4,7 +4,6 @@ from pathlib import Path from loopx.capabilities.periodic_report.pending_intent import ( - consume_pending_periodic_report_intent, pending_periodic_report_intents, ) from settlement_capability_dispatch_fixture import ( @@ -18,13 +17,7 @@ def _write_unclaimed_frontier_state(project: Path) -> None: - """State whose progress snapshot has no agent-claimed items yet. - - The unclaimed advancement Todo keeps the successor frontier owned, so the - dispatch still freezes one stage receipt, while the durable progress - snapshot stays empty and the pending intent carries capability evidence - without a frozen project_progress projection. - """ + """Leave an active successor after the earlier stage has been superseded.""" project.joinpath("goal.md").write_text( f"""# Goal @@ -70,20 +63,7 @@ def _intent_sidecar(runtime: Path) -> dict[str, object]: return json.loads(sidecars[0].read_text(encoding="utf-8")) -def _editorial_fact_sources(registry: Path, runtime: Path) -> set[str]: - result = consume_pending_periodic_report_intent( - registry_path=registry, - runtime_root=runtime, - goal_id=GOAL_ID, - agent_id=AGENT_ID, - execute=True, - ) - assert result["status"] == "editorial_required", result - request = json.loads(Path(result["editorial_request_path"]).read_text("utf-8")) - return {str(fact.get("source_ref")) for fact in request["facts"]} - - -def test_pending_intent_fallback_uses_producer_capability_evidence( +def test_later_todo_completion_does_not_replay_superseded_successor_milestone( tmp_path: Path, ) -> None: captured, registry, runtime = complete_todo_via_cli( @@ -92,23 +72,19 @@ def test_pending_intent_fallback_uses_producer_capability_evidence( write_state=_write_unclaimed_frontier_state, ) - assert "available_capabilities" in captured + assert captured["available_capabilities"] == ["network"] intents = pending_periodic_report_intents( registry_path=registry, runtime_root=runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, ) - assert len(intents) == 1 - payload = intents[0]["payload"] - assert "project_progress" not in payload - assert payload["available_capabilities"] == ["network"] - - _claim_gated_successors(registry.parent) - assert f"todo:{GATED_TODO}" in _editorial_fact_sources(registry, runtime) + assert intents == [] + sidecar = _intent_sidecar(runtime) + assert sidecar["intent"] is None -def test_pending_intent_fallback_fails_closed_without_producer_evidence( +def test_later_todo_completion_without_capabilities_does_not_replay_milestone( tmp_path: Path, ) -> None: _captured, registry, runtime = complete_todo_via_cli( @@ -117,13 +93,13 @@ def test_pending_intent_fallback_fails_closed_without_producer_evidence( write_state=_write_unclaimed_frontier_state, ) - sidecar = _intent_sidecar(runtime) - assert "project_progress" not in sidecar["intent"]["payload"] - assert "available_capabilities" not in sidecar["intent"]["payload"] - - _claim_gated_successors(registry.parent) - assert f"todo:{GATED_TODO}" not in _editorial_fact_sources(registry, runtime) - assert f"todo:{PLAIN_TODO}" in _editorial_fact_sources(registry, runtime) + assert _intent_sidecar(runtime)["intent"] is None + assert pending_periodic_report_intents( + registry_path=registry, + runtime_root=runtime, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + ) == [] def test_post_writeback_frontier_and_progress_share_one_canonical_snapshot( @@ -175,8 +151,8 @@ def observe(**kwargs): agent_id=AGENT_ID, ) assert len(reads) == 1 - assert result["stage_completion"]["acceptance"] == "validated" - assert f"todo:{GATED_TODO}" in { - r["source_ref"] for r in result["project_progress"]["items"] - } + # The history contains a later active successor, while this ordinary + # post-writeback payload has no exact refresh identity. Canonical Todo + # loading still occurs once, but it cannot recover the older milestone. + assert result == {} assert not state.exists() diff --git a/tests/cli_commands/test_todo_complete_settlement_capabilities.py b/tests/cli_commands/test_todo_complete_settlement_capabilities.py index c4ce0e9554..f434eaa39c 100644 --- a/tests/cli_commands/test_todo_complete_settlement_capabilities.py +++ b/tests/cli_commands/test_todo_complete_settlement_capabilities.py @@ -49,20 +49,7 @@ def _write_stage_state(project: Path) -> None: ) -def _dispatched_next_action_source(captured: dict[str, object]) -> str: - intents = captured["post_writeback_hooks"]["intents"] - assert isinstance(intents, list) and len(intents) == 1 - items = intents[0]["payload"]["project_progress"]["items"] - matched = [ - item["source_ref"] - for item in items - if item.get("content_kind") == "next_action" - ] - assert len(matched) == 1 - return matched[0] - - -def test_todo_complete_dispatch_selects_gated_successor_with_settlement_evidence( +def test_todo_complete_does_not_replay_superseded_successor_milestone( tmp_path: Path, ) -> None: captured, _registry, _runtime = complete_todo_via_cli( @@ -72,10 +59,11 @@ def test_todo_complete_dispatch_selects_gated_successor_with_settlement_evidence ) assert captured["available_capabilities"] == ["network"] - assert _dispatched_next_action_source(captured) == f"todo:{GATED_TODO}" + assert captured["post_writeback_hooks"]["intent_count"] == 0 + assert captured["post_writeback_hooks"]["intents"] == [] -def test_todo_complete_dispatch_fails_closed_without_capability_evidence( +def test_todo_complete_without_capabilities_does_not_replay_milestone( tmp_path: Path, ) -> None: captured, _registry, _runtime = complete_todo_via_cli( @@ -85,7 +73,8 @@ def test_todo_complete_dispatch_fails_closed_without_capability_evidence( ) assert "available_capabilities" not in captured - assert _dispatched_next_action_source(captured) == f"todo:{PLAIN_TODO}" + assert captured["post_writeback_hooks"]["intent_count"] == 0 + assert captured["post_writeback_hooks"]["intents"] == [] def _write_source_registry(path: Path, runtime: Path) -> None: @@ -185,7 +174,7 @@ def hold_lifetime_lock() -> None: assert replay["changed"] is False assert replay["completion_receipt_id"] == completion_receipt_id assert Path(str(replay["state_file"])).read_bytes() == committed_state - assert replay["post_writeback_hooks"]["intent_count"] == 1 + assert replay["post_writeback_hooks"]["intent_count"] == 0 def test_todo_complete_isolates_admission_codec_failure_and_recovers_on_replay( @@ -241,7 +230,7 @@ def for_plan(**kwargs: object) -> FirstPartyHostGoalAdmission: assert replay["changed"] is False assert replay["completion_receipt_id"] == completion_receipt_id assert Path(str(replay["state_file"])).read_bytes() == committed_state - assert replay["post_writeback_hooks"]["intent_count"] == 1 + assert replay["post_writeback_hooks"]["intent_count"] == 0 def test_todo_complete_isolates_lifetime_transport_failure(