fix(collaboration): honor canonical stopped goals

loopx/chat_loopx_mode.py

@@ -16,6 +16,9 @@
 
 from .agent_registry import load_goal_from_registry, registered_agent_ids_for_goal
 from .chat_codex_goal import CodexGoalDriver, validate_goal_chat
+from .control_plane.collaboration.goal_instance_scope import (
+    goal_accepts_collaboration,
+)
 from .control_plane.effect_runtime import effect_runtime_result
 from .control_plane.collaboration.inbox import _read, _root
 from .file_lock import exclusive_file_lock, LockAcquisitionPolicy, LockAcquireTimeoutError
@@ -420,7 +423,7 @@ def apply(self, session_id, body, *, work_dir, objective):
                     "settings": settings,
                     "native": session.get("native_goal") or {},
                     "registered_agents": registered_agent_ids_for_goal(goal),
-                    "goal_active": goal.get("status") not in {"stopped", "archived"},
+                    "goal_active": goal_accepts_collaboration(goal),
                     "execution_binding_valid": True,
                 },
             )
@@ -534,7 +537,7 @@ def settled(state, reason, **facts):
                 "native": session.get("native_goal") or {},
                 "registered_agents": registered_agent_ids_for_goal(goal) if goal else [],
                 "goal_active": goal is not None
-                and goal.get("status") not in {"stopped", "archived"},
+                and goal_accepts_collaboration(goal),
                 "execution_binding_valid": binding_valid,
             },
         )

loopx/control_plane/collaboration/goal_instance_scope.py

@@ -9,6 +9,7 @@
 from ...agent_registry import registered_agent_ids_for_goal
 from ...file_lock import exclusive_cross_runtime_file_lock
 from ..effect_runtime import effect_runtime_result
+from ..goals.activation import goal_is_stopped
 from ..goals.source_session_registry_state import exact_goal_ref, guard_path
 from ..projects.registry_codec import (
     SOURCE_SESSION_PROFILE_ID,
@@ -44,6 +45,13 @@ def record_identity(self) -> dict[str, Any]:
         return result
 
 
+def goal_accepts_collaboration(goal: dict[str, Any]) -> bool:
+    return not goal_is_stopped(goal) and goal.get("status") not in {
+        "stopped",
+        "archived",
+    }
+
+
 def _registered_goal(
     registry: dict[str, Any],
     *,
@@ -63,7 +71,7 @@ def _registered_goal(
         agent not in registered_agent_ids_for_goal(goal) for agent in agents
     ):
         raise ValueError("collaboration requires registered Agents of the same Goal")
-    if require_active and goal.get("status") in {"stopped", "archived"}:
+    if require_active and not goal_accepts_collaboration(goal):
         raise ValueError("collaboration Goal is stopped or archived")
     return goal
 

loopx/control_plane/collaboration/peers.py

@@ -29,6 +29,7 @@
 from .goal_instance_scope import (
     collaboration_goal_scope,
     decide_collaboration_lifecycle,
+    goal_accepts_collaboration,
 )
 from ...agent_registry import registered_agent_ids_for_goal
 from ...thread_agent_binding import resolve_thread_agent_binding
@@ -57,7 +58,7 @@ def _goal(registry, goal_id, *agents, require_active=False):
     )
     if not goal or any(a not in registered_agent_ids_for_goal(goal) for a in agents):
         raise ValueError("peer request requires registered Agents of the same Goal")
-    if require_active and goal.get("status") in {"stopped", "archived"}:
+    if require_active and not goal_accepts_collaboration(goal):
         raise ValueError("peer request Goal is stopped or archived")
     return goal
 

loopx/semantics/project_registry_io_manifest_v1.json

@@ -1031,7 +1031,7 @@
     },
     {
       "site": "loopx/control_plane/collaboration/goal_instance_scope.py::<module>.collaboration_goal_scope::codec_read:load_project_registry#1",
-      "line": 121,
+      "line": 129,
       "column": 20,
       "kind": "codec_read",
       "api": "load_project_registry",
@@ -1047,7 +1047,7 @@
     },
     {
       "site": "loopx/control_plane/collaboration/peers.py::<module>._goal::codec_read:load_project_registry#1",
-      "line": 53,
+      "line": 54,
       "column": 22,
       "kind": "codec_read",
       "api": "load_project_registry",

skills/loopx-self-repair/references/repair-patterns.md

@@ -16,6 +16,7 @@ teaches a reusable control-plane lesson.
 | `acceptance_validation_failure_flattened` | A bound Todo is `ready` and its ordinary validator passes, yet completion reports only `goal_acceptance_validation_rejected`; the agent searches contract bindings before discovering a workspace or runner failure. | Exact Todo acceptance state, completion's typed criterion failure, recorded delivery workspace, current worktree cleanliness, and the privacy-safe runner receipt. | The completion boundary collapsed a failed criterion receipt into a generic contract error, hiding the workspace or command status. | Project the failed criterion ID, allowlisted validation status, safe exit code and bounded next action without command output or local paths. Repair the execution context and retry under the same Turn/lease; do not rebind owner criteria or infer a Goal-wide hold. |
 | `review_compatibility_assumption_gap` | A correct fix retains parallel protocol paths, and the review treats historical receipt recovery as proof that every old request decoder is needed. | Published review, structured compatibility rationale, real caller/deployment inventory, stored request versus receipt shape, and smaller-design readback. | Re-review verifies the last bug but does not separate compatibility obligations or test consolidation; free-text claims pass as evidence. | Replace the capability's existing compatibility rationale with a bounded structured assessment. Distinguish transient requests, independent client rollout and persisted replay formats; compare one typed current contract; preserve real legacy consumers. Cover needless retention and unsafe removal with positive twins. Keep optional simplifications advisory and do not make field completeness certify evidence truth. |
 | `review_outcome_continuity_gap` | Local feature checks pass while later work starves, recovery only records a blocker, or ordinary users face new repeated intervention. | Accepted product outcome, published review, real later invocations, affected user surfaces, durable progress and authorized recovery. | Review proved one operation or prevention of bypass without checking sustained progress and the complete user journey. | Make long-horizon continuity and user experience explicit judgments in the existing problem context. Reuse bounded real walkthroughs; apply scope counterexamples when gates are involved. Accept deliberate waits only with an independent basis and recovery or safe terminal route. Qualify with exact-source historical counterexamples and repaired positive controls; keep old verdicts out of model inputs and distinguish reported historical checks from newly executed tests. Section length and declared evidence cannot certify truth. For frontend friction, compare the shortest authorized task path before encoding the UI sequence; inspect populated composition and real state readback. Record fixture/build/install boundaries and repair the existing delivery workflow, not a second checklist. |
+| `goal_activation_consumer_drift` | A canonically stopped Goal still accepts new peer requests, delegations, Chat execution, scheduled work, or another effect. | The registry Goal's `activation` object, any legacy top-level `status`, the active-admission predicate at every effect entrypoint, and state bytes or process calls before refusal. | The activation writer moved to the typed `activation.state` owner, but one or more consumers still read only the legacy top-level `status`. | Reuse `goal_activation_state()` or `goal_is_stopped()` in one domain predicate, retain only required legacy or archival compatibility, and route every effect entrypoint through that predicate. Run red-green tests for the canonical nested state and assert refusal occurs before writes, process launch, or external effects. Keep historical reads available after stop. |
 | `source_context_legacy_downgrade` | An unstamped record mutates state through a legacy path after the authoritative source disappears, although another record in the same owner context preserves an exact source identity. | Selected record identity, all same-context durable candidates, source path availability, operation entrypoint, and business-file bytes before and after refusal. | The adapter treated one selected record's missing identity as proof that the whole context was legacy and ignored exact historical evidence when the current authority could not be read. | Resolve source-context availability once for selection and mutation. Exact history may prove that legacy fallback is unsafe, but it must never supply or guess the current identity. Fail closed before writes when authority is unavailable; preserve contexts with no exact history through opposite legacy controls across every mutation entrypoint. |
 | `archive_capture_classification_gap` | Whole-Goal capture rejects a reachable archived Agent record although its active read was valid. | Recorded role/class, existing legacy read classification, transitive dependency closure, bootstrap and writer-outbox readback. | Archive storage preserved the role but omitted the resolved class; capture treated missing class as missing authority. | Keep recorded identity separate from compatibility classification. Only a recorded Agent role can adopt the existing read class; use it consistently for closure and materialization. Preserve the class on new archive moves, keep user authority fail-closed, and validate full source capture in disposable real providers without changing the active Goal. |
 | `shadow_proof_transport_amplification` | A large Goal cannot capture its first mutation or finish drain although the same tiny Goal succeeds; RPC rejects an oversized response. | Same source population, base/head serialized response sizes, actual sequence/drain callers, qualified lineage and cursor readback. | A consumer needing progress or partition markers received the full head and every historical projection across the language boundary. | Keep full history verification in the typed owner and return a purpose-specific compact proof. Preserve receipts, sequence and lineage checks; do not raise transport limits, truncate source records or weaken qualification to make the test pass. Cover the old oversized response and real CLI capture, drain and reviewed cutover on a disposable snapshot. |

tests/fixtures/self_repair_queries.json

@@ -8,5 +8,6 @@
   {"query": "CLI still old quota behavior after merge", "relevant": "local_release_snapshot_stale_after_merge"},
   {"query": "shadow capture oversized RPC response", "relevant": "shadow_proof_transport_amplification"},
   {"query": "review retains duplicate protocol compatibility decoders", "relevant": "review_compatibility_assumption_gap"},
-  {"query": "contract criteria runner failed workspace completion", "relevant": "acceptance_validation_failure_flattened"}
+  {"query": "contract criteria runner failed workspace completion", "relevant": "acceptance_validation_failure_flattened"},
+  {"query": "canonical activation stopped peer delegation still accepted", "relevant": "goal_activation_consumer_drift"}
 ]

tests/test_chat_delegation_wake.py

@@ -214,12 +214,23 @@ def test_paused_lead_stays_pending_and_is_not_unpaused(mode):  # noqa: F811
     assert calls == []
 
 
-def test_stopped_goal_refuses_the_wake(mode):  # noqa: F811
+@pytest.mark.parametrize("state_format", ["legacy", "canonical"])
+def test_stopped_goal_refuses_the_wake(mode, state_format):  # noqa: F811
     service, sid, calls = _idle_resumable_lead(mode)
     repo = mode[2]
     registry = service.controller.registry_path
     payload = json.loads(registry.read_text())
-    next(goal for goal in payload["goals"] if goal["id"] == "research")["status"] = "stopped"
+    goal = next(goal for goal in payload["goals"] if goal["id"] == "research")
+    if state_format == "legacy":
+        goal["status"] = "stopped"
+    else:
+        goal.pop("status", None)
+        goal["activation"] = {
+            "schema_version": "loopx_goal_activation_v1",
+            "state": "stopped",
+            "updated_at": "2026-10-06T00:00:00Z",
+            "reason": "Owner stopped the Goal.",
+        }
     registry.write_text(json.dumps(payload))
     path = _write_record(service, session_id=sid)
 

tests/test_chat_loopx_mode.py

@@ -9,7 +9,7 @@
 from loopx.chat_loopx_mode import TOOL
 from loopx.chat_runtime import ChatRuntimeController
 from loopx.chat_store import ChatSessionStore
-from loopx.control_plane.effect_runtime import effect_runtime_result
+from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result
 from test_chat_project_coordination import project  # noqa: F401
 
 
@@ -151,6 +151,26 @@ def test_goal_registry_is_the_execution_config_owner(mode):
         )
 
 
+def test_canonical_stopped_goal_rejects_start_before_turn_creation(mode):
+    service, _, _, settings, calls = mode
+    registry = service.controller.registry_path
+    payload = json.loads(registry.read_text())
+    goal = next(item for item in payload["goals"] if item["id"] == "research")
+    goal.pop("status", None)
+    goal["activation"] = {
+        "schema_version": "loopx_goal_activation_v1",
+        "state": "stopped",
+        "updated_at": "2026-10-06T00:00:00Z",
+        "reason": "Owner stopped the Goal.",
+    }
+    registry.write_text(json.dumps(payload))
+
+    with pytest.raises(EffectRuntimeRejected, match="stopped"):
+        apply(mode, "start", settings=settings)
+
+    assert calls == []
+
+
 def test_unfinished_legacy_session_can_resume_until_goal_config_is_migrated(mode):
     service, sid, repo, settings, calls = mode
     registry = service.controller.registry_path

tests/test_local_delegation.py

@@ -204,6 +204,30 @@ def test_delegation_captures_goal_ref_after_registry_becomes_available(tmp_path,
     assert len(calls) == 1
 
 
+def test_canonical_stopped_goal_rejects_new_delegation_before_launch(
+    service, monkeypatch
+):
+    _, runner = service
+    registry = json.loads(runner.registry.read_text())
+    goal = registry["goals"][0]
+    goal.pop("status", None)
+    goal["activation"] = {
+        "schema_version": "loopx_goal_activation_v1",
+        "state": "stopped",
+        "updated_at": "2026-10-06T00:00:00Z",
+        "reason": "Owner stopped the Goal.",
+    }
+    runner.registry.write_text(json.dumps(registry))
+    launched = []
+    monkeypatch.setattr(runner, "_spawn", launched.append)
+
+    with pytest.raises(ValueError, match="stopped"):
+        runner.start("analysis", "canonical-stopped", brief())
+
+    assert launched == []
+    assert not runner.path("canonical-stopped").exists()
+
+
 def test_worker_waits_for_a_transient_status_probe(service, monkeypatch):
     """A reader temporarily holding the lock must not discard admitted work."""
     from loopx import collaboration_mcp as delegation

tests/test_peer_collaboration.py

@@ -623,6 +623,41 @@ def test_special_file_read_is_bounded_and_stopped_goal_remains_readable(scenario
         request(root, registry, "delivery", "builder", "reviewer", "stopped", brief)
 
 
+def test_canonical_stopped_goal_rejects_peer_request_before_any_write(scenario):
+    root, registry, brief, *_ = scenario
+    config = json.loads(registry.read_text())
+    goal = config["goals"][0]
+    goal.pop("status", None)
+    goal["activation"] = {
+        "schema_version": "loopx_goal_activation_v1",
+        "state": "stopped",
+        "updated_at": "2026-10-06T00:00:00Z",
+        "reason": "Owner stopped the Goal.",
+    }
+    registry.write_text(json.dumps(config))
+    state_root = root / ".local/manager-context"
+    before = {
+        path.relative_to(state_root): path.read_bytes()
+        for path in state_root.rglob("*.json")
+    }
+
+    with pytest.raises(ValueError, match="stopped"):
+        request(
+            root,
+            registry,
+            "delivery",
+            "builder",
+            "reviewer",
+            "canonical-stopped",
+            brief,
+        )
+
+    assert {
+        path.relative_to(state_root): path.read_bytes()
+        for path in state_root.rglob("*.json")
+    } == before
+
+
 def test_nested_coordinators_return_to_each_immediate_requester(scenario):
     """Worker -> coordinator -> specialist works without a manager root request."""
     root, registry, brief, *_ = scenario