fix(periodic-report): honor canonical Goal stops

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",

loopx/capabilities/periodic_report/cadence_runtime.py

@@ -13,7 +13,8 @@
 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
 
@@ -25,11 +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.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.
@@ -50,7 +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 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:
@@ -62,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),

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",
 ]

loopx/capabilities/periodic_report/pending_intent.py

@@ -43,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
@@ -231,13 +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.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(

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):

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,19 +266,39 @@ 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)
     kwargs = dict(registry_path=registry, runtime_root=runtime, goal_id=GOAL_ID, agent_id=AGENT_ID)
     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

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: