From d9fa9a68a69cc9f2dfaa7863c337f65fd14d96bc Mon Sep 17 00:00:00 2001 From: duang777 Date: Thu, 15 Oct 2026 23:42:17 +0800 Subject: [PATCH 1/2] fix(periodic-report): honor canonical Goal stops Signed-off-by: duang777 --- .../periodic_report/cadence_runtime.py | 13 +++++- .../periodic_report/pending_intent.py | 7 +++- .../test_periodic_report_pending_intent.py | 41 ++++++++++++++++--- 3 files changed, 52 insertions(+), 9 deletions(-) diff --git a/loopx/capabilities/periodic_report/cadence_runtime.py b/loopx/capabilities/periodic_report/cadence_runtime.py index 349d9cafc4..d0deecb0ca 100644 --- a/loopx/capabilities/periodic_report/cadence_runtime.py +++ b/loopx/capabilities/periodic_report/cadence_runtime.py @@ -9,6 +9,7 @@ from ...control_plane.capability_hooks import ( TURN_START_HOOK_RESULT_SCHEMA_VERSION, TurnStartHookRegistration, dispatch_turn_start_hooks, ) +from ...control_plane.goals.activation import goal_is_stopped from ...history import load_registry from ...registry import find_registry_goal from .cadence_journal import admit_cadence_window, read_cadence_journal @@ -25,7 +26,11 @@ def periodic_report_cadence_hooks( if not agent_id: return () goal = find_registry_goal(load_registry(registry_path), goal_id) - if not isinstance(goal, dict) or goal.get("status") in {"stopped", "paused", "archived"}: + if ( + not isinstance(goal, dict) + or goal_is_stopped(goal) + or goal.get("status") in {"stopped", "paused", "archived"} + ): return () subscription = resolve_goal_periodic_report_subscription( goal, read_periodic_report_machine_defaults(runtime_root)) @@ -50,7 +55,11 @@ def produce(): def resolve_current_subscription(): nonlocal current_reporters current_goal = find_registry_goal(load_registry(registry_path), goal_id) - if not isinstance(current_goal, dict) or current_goal.get("status") in {"stopped", "paused", "archived"}: + if ( + not isinstance(current_goal, dict) + or goal_is_stopped(current_goal) + or current_goal.get("status") in {"stopped", "paused", "archived"} + ): return None current_reporters = registered_agent_ids_for_goal(current_goal) if not current_reporters or agent_id not in current_reporters: diff --git a/loopx/capabilities/periodic_report/pending_intent.py b/loopx/capabilities/periodic_report/pending_intent.py index 5a63c19179..a12ecef593 100644 --- a/loopx/capabilities/periodic_report/pending_intent.py +++ b/loopx/capabilities/periodic_report/pending_intent.py @@ -17,6 +17,7 @@ ) from ...control_plane.effect_runtime import effect_runtime_result from ...control_plane.digest_envelope import sha256_envelope +from ...control_plane.goals.activation import goal_is_stopped from ...history import load_registry from .todo_source import read_report_todo_source from ...registry import ( @@ -231,7 +232,11 @@ def _active_delivery_subscription( registry = load_registry(registry_path) goal = find_registry_goal(registry, goal_id) - if not isinstance(goal, Mapping) or goal.get("status") in {"stopped", "paused", "archived"}: + if ( + not isinstance(goal, Mapping) + or goal_is_stopped(goal) + or goal.get("status") in {"stopped", "paused", "archived"} + ): return None subscription = resolve_goal_periodic_report_subscription( goal, diff --git a/tests/capabilities/test_periodic_report_pending_intent.py b/tests/capabilities/test_periodic_report_pending_intent.py index 1f0e97f1e5..24b8e14f87 100644 --- a/tests/capabilities/test_periodic_report_pending_intent.py +++ b/tests/capabilities/test_periodic_report_pending_intent.py @@ -222,7 +222,9 @@ def test_calendar_reporter_replacement_uses_admitted_window_state(tmp_path, prep assert current["publication"] is None -@pytest.mark.parametrize("change", ["reporter", "schedule", "disabled", "stopped"]) +@pytest.mark.parametrize( + "change", ["reporter", "schedule", "disabled", "stopped", "activation_stopped"] +) def test_calendar_hook_rechecks_live_authority_after_construction(tmp_path, change): from datetime import datetime from loopx.control_plane.capability_hooks import dispatch_turn_start_hooks @@ -241,8 +243,15 @@ def test_calendar_hook_rechecks_live_authority_after_construction(tmp_path, chan goal["control_plane"]["periodic_report"]["schedule"]["rrule"] = "FREQ=WEEKLY;BYDAY=FR;BYHOUR=17" elif change == "disabled": goal["control_plane"]["periodic_report"]["enabled"] = False - else: + elif change == "stopped": goal["status"] = "stopped" + else: + goal["activation"] = { + "schema_version": "loopx_goal_activation_v1", + "state": "stopped", + "updated_at": "2026-09-11T10:00:00Z", + "reason": "Owner stopped automatic work.", + } registry.write_text(json.dumps(config)) dispatched = dispatch_turn_start_hooks(hooks) assert not dispatched["failures"] @@ -257,10 +266,29 @@ def test_calendar_hook_rechecks_live_authority_after_construction(tmp_path, chan assert read_cadence_journal(runtime_root=runtime, goal_id=GOAL_ID)["window"]["agent_id"] == "replacement-peer" -@pytest.mark.parametrize("status", ["stopped", "paused", "archived"]) -def test_inactive_goal_does_not_reactivate_an_admitted_report(tmp_path, status): +@pytest.mark.parametrize( + ("field", "value"), + [ + ("status", "stopped"), + ("status", "paused"), + ("status", "archived"), + ( + "activation", + { + "schema_version": "loopx_goal_activation_v1", + "state": "stopped", + "updated_at": "2026-09-11T10:00:00Z", + "reason": "Owner stopped automatic work.", + }, + ), + ], +) +def test_inactive_goal_does_not_reactivate_an_admitted_report(tmp_path, field, value): from datetime import datetime - from loopx.capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch + from loopx.capabilities.periodic_report.cadence_runtime import ( + extend_cadence_turn_start_dispatch, + periodic_report_cadence_hooks, + ) from loopx.capabilities.periodic_report.cadence_journal import read_cadence_journal registry, runtime = _calendar_fixture(tmp_path) @@ -268,8 +296,9 @@ def test_inactive_goal_does_not_reactivate_an_admitted_report(tmp_path, status): extend_cadence_turn_start_dispatch({}, **kwargs, now=datetime.fromisoformat("2026-09-11T10:00:00+00:00")) original = read_cadence_journal(runtime_root=runtime, goal_id=GOAL_ID) config = json.loads(registry.read_text()) - config["goals"][0]["status"] = status + config["goals"][0][field] = value registry.write_text(json.dumps(config)) + assert periodic_report_cadence_hooks(**kwargs) == () assert pending_periodic_report_intents(**kwargs) == [] assert consume_pending_periodic_report_intent(**kwargs, execute=True)["status"] == "no_pending_intent" assert read_cadence_journal(runtime_root=runtime, goal_id=GOAL_ID) == original From 3a9c00391e0c31b5575a5400ea4ecd432605e7b4 Mon Sep 17 00:00:00 2001 From: duang777 Date: Fri, 16 Oct 2026 10:37:55 +0800 Subject: [PATCH 2/2] fix(periodic-report): stop queued delivery Signed-off-by: duang777 --- .../capabilities/periodic_report/__init__.py | 2 + .../periodic_report/cadence_runtime.py | 25 ++---- .../periodic_report/machine_defaults.py | 18 +++++ .../periodic_report/pending_intent.py | 13 +-- .../lark/periodic_report_delivery.py | 6 +- ...t_periodic_report_goal_channel_delivery.py | 80 +++++++++++++++++++ 6 files changed, 114 insertions(+), 30 deletions(-) diff --git a/loopx/capabilities/periodic_report/__init__.py b/loopx/capabilities/periodic_report/__init__.py index 325f13be50..21eb7f4533 100644 --- a/loopx/capabilities/periodic_report/__init__.py +++ b/loopx/capabilities/periodic_report/__init__.py @@ -31,6 +31,7 @@ "normalize_periodic_report_machine_defaults": "machine_defaults", "normalize_periodic_report_delivery_authority": "machine_defaults", "periodic_report_machine_configuration_namespace": "machine_defaults", + "resolve_active_goal_periodic_report_subscription": "machine_defaults", "resolve_goal_periodic_report_subscription": "machine_defaults", "select_goal_periodic_report_executor": "machine_defaults", "PERIODIC_REPORT_PROFILE_PRESET_ALIASES": "presets", @@ -95,6 +96,7 @@ def __dir__() -> list[str]: "normalize_periodic_report_trigger_policy", "project_progress_periodic_report_source_adapter", "resolve_periodic_report_profile_preset", + "resolve_active_goal_periodic_report_subscription", "resolve_goal_periodic_report_subscription", "select_goal_periodic_report_executor", "verify_periodic_report_archive_receipts", diff --git a/loopx/capabilities/periodic_report/cadence_runtime.py b/loopx/capabilities/periodic_report/cadence_runtime.py index d0deecb0ca..0c500baff7 100644 --- a/loopx/capabilities/periodic_report/cadence_runtime.py +++ b/loopx/capabilities/periodic_report/cadence_runtime.py @@ -9,12 +9,12 @@ from ...control_plane.capability_hooks import ( TURN_START_HOOK_RESULT_SCHEMA_VERSION, TurnStartHookRegistration, dispatch_turn_start_hooks, ) -from ...control_plane.goals.activation import goal_is_stopped from ...history import load_registry from ...registry import find_registry_goal from .cadence_journal import admit_cadence_window, read_cadence_journal from .machine_defaults import ( - resolve_goal_periodic_report_subscription, select_goal_periodic_report_executor, + resolve_active_goal_periodic_report_subscription, + select_goal_periodic_report_executor, ) from .machine_store import read_periodic_report_machine_defaults @@ -26,15 +26,9 @@ def periodic_report_cadence_hooks( if not agent_id: return () goal = find_registry_goal(load_registry(registry_path), goal_id) - if ( - not isinstance(goal, dict) - or goal_is_stopped(goal) - or goal.get("status") in {"stopped", "paused", "archived"} - ): - return () - subscription = resolve_goal_periodic_report_subscription( + subscription = resolve_active_goal_periodic_report_subscription( goal, read_periodic_report_machine_defaults(runtime_root)) - if subscription.get("enabled") is not True or subscription.get("schedule") is None: + if subscription is None or subscription.get("schedule") is None: return () # Prefer the frozen reporter for unfinished work. Registration reordering # must not make its recovery hook disappear or elect a second reporter. @@ -55,11 +49,9 @@ def produce(): def resolve_current_subscription(): nonlocal current_reporters current_goal = find_registry_goal(load_registry(registry_path), goal_id) - if ( - not isinstance(current_goal, dict) - or goal_is_stopped(current_goal) - or current_goal.get("status") in {"stopped", "paused", "archived"} - ): + current_subscription = resolve_active_goal_periodic_report_subscription( + current_goal, read_periodic_report_machine_defaults(runtime_root)) + if current_subscription is None: return None current_reporters = registered_agent_ids_for_goal(current_goal) if not current_reporters or agent_id not in current_reporters: @@ -71,8 +63,7 @@ def resolve_current_subscription(): reporting_agent_id=reporter, eligible_agent_ids=current_reporters) if selected["selected_agent_id"] != agent_id: return None - return resolve_goal_periodic_report_subscription( - current_goal, read_periodic_report_machine_defaults(runtime_root)) + return current_subscription admission = admit_cadence_window(runtime_root=runtime_root, goal_id=goal_id, agent_id=agent_id, subscription=subscription, now=now or datetime.now(timezone.utc), diff --git a/loopx/capabilities/periodic_report/machine_defaults.py b/loopx/capabilities/periodic_report/machine_defaults.py index c34632d0f3..729c681055 100644 --- a/loopx/capabilities/periodic_report/machine_defaults.py +++ b/loopx/capabilities/periodic_report/machine_defaults.py @@ -9,6 +9,7 @@ from .cadence import normalize_report_cadence from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN +from ...control_plane.goals.activation import goal_is_stopped from ...control_plane.todos.contract import normalize_todo_claimed_by from ...control_plane.digest_envelope import sha256_envelope from ..configuration_ui import resolve_capability_configuration @@ -372,6 +373,22 @@ def resolve_goal_periodic_report_subscription( ) from exc +def resolve_active_goal_periodic_report_subscription( + goal: Mapping[str, Any] | None, + machine_defaults: Mapping[str, Any] | None = None, +) -> dict[str, Any] | None: + """Resolve the current standing authority for one Goal subscription.""" + + if ( + not isinstance(goal, Mapping) + or goal_is_stopped(goal) + or goal.get("status") in {"stopped", "paused", "archived"} + ): + return None + subscription = resolve_goal_periodic_report_subscription(goal, machine_defaults) + return subscription if subscription.get("enabled") is True else None + + def normalize_periodic_report_delivery_authority(raw: object) -> dict[str, Any]: """Validate the frozen standing authority carried to an external delivery.""" @@ -580,6 +597,7 @@ def build_goal_periodic_report_delivery_plan( "normalize_periodic_report_delivery_authority", "normalize_periodic_report_machine_defaults", "periodic_report_machine_configuration_namespace", + "resolve_active_goal_periodic_report_subscription", "resolve_goal_periodic_report_subscription", "select_goal_periodic_report_executor", ] diff --git a/loopx/capabilities/periodic_report/pending_intent.py b/loopx/capabilities/periodic_report/pending_intent.py index a12ecef593..6d85348735 100644 --- a/loopx/capabilities/periodic_report/pending_intent.py +++ b/loopx/capabilities/periodic_report/pending_intent.py @@ -17,7 +17,6 @@ ) from ...control_plane.effect_runtime import effect_runtime_result from ...control_plane.digest_envelope import sha256_envelope -from ...control_plane.goals.activation import goal_is_stopped from ...history import load_registry from .todo_source import read_report_todo_source from ...registry import ( @@ -44,7 +43,7 @@ ) from .machine_defaults import ( build_periodic_report_delivery_authority, - resolve_goal_periodic_report_subscription, + resolve_active_goal_periodic_report_subscription, ) from .machine_store import read_periodic_report_machine_defaults from .cadence_journal import cadence_intent, cadence_journal_path, read_cadence_journal @@ -232,17 +231,11 @@ def _active_delivery_subscription( registry = load_registry(registry_path) goal = find_registry_goal(registry, goal_id) - if ( - not isinstance(goal, Mapping) - or goal_is_stopped(goal) - or goal.get("status") in {"stopped", "paused", "archived"} - ): - return None - subscription = resolve_goal_periodic_report_subscription( + subscription = resolve_active_goal_periodic_report_subscription( goal, read_periodic_report_machine_defaults(runtime_root), ) - return dict(subscription) if subscription.get("enabled") is True else None + return dict(subscription) if subscription is not None else None def _next_attempt_revision( diff --git a/loopx/extensions/lark/periodic_report_delivery.py b/loopx/extensions/lark/periodic_report_delivery.py index 7025b9f7e4..8c384c0571 100644 --- a/loopx/extensions/lark/periodic_report_delivery.py +++ b/loopx/extensions/lark/periodic_report_delivery.py @@ -21,7 +21,7 @@ from ...capabilities.periodic_report.machine_defaults import ( build_periodic_report_delivery_authority, normalize_periodic_report_delivery_authority, - resolve_goal_periodic_report_subscription, + resolve_active_goal_periodic_report_subscription, ) from ...capabilities.periodic_report.machine_store import ( read_periodic_report_machine_defaults, @@ -195,11 +195,11 @@ def _resolved_goal_channel_binding( ) -> dict[str, Any]: registry = load_registry(registry_path) goal = goal_from_registry(registry, goal_id) - subscription = resolve_goal_periodic_report_subscription( + subscription = resolve_active_goal_periodic_report_subscription( goal, read_periodic_report_machine_defaults(runtime_root), ) - if subscription.get("enabled") is not True: + if subscription is None: raise ValueError("periodic report delivery subscription is disabled") current_authority = build_periodic_report_delivery_authority(subscription) if current_authority != dict(expected_authority): diff --git a/tests/extensions/test_periodic_report_goal_channel_delivery.py b/tests/extensions/test_periodic_report_goal_channel_delivery.py index 5b66b7eca6..6566498b65 100644 --- a/tests/extensions/test_periodic_report_goal_channel_delivery.py +++ b/tests/extensions/test_periodic_report_goal_channel_delivery.py @@ -868,6 +868,30 @@ def test_goal_channel_delivery_revalidates_subscription_before_explicit_binding_ assert calls == [] +def test_goal_channel_delivery_rejects_canonical_stop_before_send( + tmp_path: Path, +) -> None: + registry_path = tmp_path / ".loopx" / "registry.json" + stopped = _goal() + stopped["activation_state"] = "stopped" + _write_registry(registry_path, goal=stopped) + _write_binding(registry_path) + calls: list[list[str]] = [] + + with pytest.raises(ValueError, match="subscription is disabled"): + deliver_periodic_report_to_goal_channel( + _request(), + registry_path=registry_path, + runtime_root=tmp_path / "runtime", + goal_id=GOAL_ID, + extension_activation=_extension_activation(), + execute=True, + runner=_runner(calls), + ) + + assert not any("+messages-send" in args for args in calls) + + def test_goal_channel_delivery_revalidates_live_machine_default_before_send( tmp_path: Path, ) -> None: @@ -937,6 +961,62 @@ def runner( assert len([args for args in calls if "+messages-send" in args]) == 1 +def test_goal_channel_delivery_stops_between_writes_then_resumes_without_resend( + tmp_path: Path, +) -> None: + registry_path = tmp_path / ".loopx" / "registry.json" + _write_registry(registry_path) + _write_binding(registry_path) + calls: list[list[str]] = [] + base_runner = _runner(calls) + stopped = False + + def runner( + args: list[str], + cwd: Path | None, + timeout: float | None, + ) -> dict[str, Any]: + nonlocal stopped + result = base_runner(args, cwd, timeout) + if "+messages-send" in args and not stopped: + stopped = True + stopped_goal = _goal() + stopped_goal["activation_state"] = "stopped" + _write_registry(registry_path, goal=stopped_goal) + return result + + with pytest.raises(ValueError, match="subscription is disabled"): + deliver_periodic_report_to_goal_channel( + _request(), + registry_path=registry_path, + runtime_root=tmp_path / "runtime", + goal_id=GOAL_ID, + extension_activation=_extension_activation(), + execute=True, + runner=runner, + ) + + assert len([args for args in calls if "+messages-send" in args]) == 1 + + _write_registry(registry_path) + resumed = deliver_periodic_report_to_goal_channel( + _request(), + registry_path=registry_path, + runtime_root=tmp_path / "runtime", + goal_id=GOAL_ID, + extension_activation=_extension_activation(), + execute=True, + runner=runner, + ) + + assert resumed["status"] == "satisfied" + assert len([args for args in calls if "+messages-send" in args]) == 2 + assert [ + item["semantic_dedupe_status"] + for item in resumed["sink_result"]["message_results"] + ] == ["existing_exact_message", "no_existing_exact_message"] + + def test_goal_channel_delivery_requires_native_message_sender_readback( tmp_path: Path, ) -> None: