Exclude stopped Goals from steward handoff recipients

docs/architecture/rfcs/capable-manager-semantic-handoff-v0.md

@@ -223,6 +223,8 @@ Do not put the entire handoff inside Goal Vision's bounded summary or expand eve
 
 Discover all registered active Goals and their agents within authorized host scope. Stopped Goals are excluded by default but can be requested. Discoverability, read access, context delivery and execution permission are four separate facts. “Best effort” means the manager investigates role, current work, repository and availability; it does not mean broadcasting private context or guessing an identity.
 
+The existing manager context-delivery catalog now excludes Goals marked stopped in its selected registry, and delivery rechecks that activation guard before creating or replaying an inbox request. A malformed activation state cannot grant a recipient. This is one admission boundary only: a lagging global mirror still needs source-authority reconciliation, and registration does not establish a live session, executable capacity, suitable model or a returned result. Explicit inspection of a stopped Goal remains separate from delivering it new work.
+
 Prefer the explicitly named receiver; otherwise resolve the best responsible agent from current state. A missing convenience routing profile must not make a known authorized worker nonexistent. Avoid fixed per-request catalogs as the only responsibility model. If several agents fit, choose an assessment owner and say why; clarify only when ambiguity materially changes authority or outcome. If no worker is suitable, do the permitted work locally or report the actual missing capability. Do not silently start a new user task or wake a stopped Goal.
 
 The receiver gets the packet at a supported safe interaction boundary. The existing turn-start hook is the baseline; an attached runtime may support prompt delivery at its next safe continuation point. Always publish whether the target is reachable, queued until wake, or unsupported. Inbox storage does not prove injection into a model session, and injection does not prove adoption.

loopx/capabilities/manager_context/__init__.py

@@ -10,6 +10,7 @@
 from ...file_lock import exclusive_file_lock
 from ...history import load_registry
 from ...control_plane.collaboration import conversation_scope
+from ...control_plane.goals.activation import goal_is_stopped
 
 # Retained imports are the shipped manager-context API; the shared owner is neutral.
 from ...control_plane.collaboration.inbox import (
@@ -75,12 +76,17 @@ def authority(
             raise ValueError("invalid registry")
     except (OSError, ValueError, TypeError):
         return {"mode": "unavailable", "targets": []}
-    available = {
-        (g["id"], a): g
-        for g in registry.get("goals", [])
-        if isinstance(g, dict) and g.get("id")
-        for a in registered_agent_ids_for_goal(g)
-    }
+    available = set()
+    for goal in registry.get("goals", []):
+        if not isinstance(goal, dict) or not goal.get("id"):
+            continue
+        try:
+            if goal_is_stopped(goal):
+                continue
+        except ValueError:
+            # An unreadable activation state cannot grant a new handoff.
+            continue
+        available.update((goal["id"], agent) for agent in registered_agent_ids_for_goal(goal))
     scope = conversation_scope(session, origin=turn.get("origin", "unknown"))
     if scope["private_conversation"] and turn.get("origin") == "web":
         allowed = {target for target in available
@@ -112,7 +118,7 @@ def authority(
         except (OSError, ValueError, KeyError, TypeError, AttributeError):
             return {"mode": "unavailable", "targets": []}
     targets = [
-        {"goal_id": g, "agent_id": a} for g, a in sorted(allowed & set(available))
+        {"goal_id": g, "agent_id": a} for g, a in sorted(allowed & available)
     ]
     return {
         "mode": "context_only",

tests/test_manager_context_handoff.py

@@ -69,6 +69,64 @@ def test_project_conversation_delivers_only_to_its_registered_goal(fixture):
     assert not pending(root, "other", "peer")["items"]
 
 
+def test_stopped_goal_is_not_a_context_recipient_and_revokes_replay(fixture):
+    root, registry, session, turn, request = fixture
+    assert request in authority(root, registry, session, turn)["targets"]
+    first = deliver(root, registry, session=session, turn=turn, request=request)
+
+    data = json.loads(registry.read_text())
+    data["goals"][0]["activation_state"] = "stopped"
+    registry.write_text(json.dumps(data))
+    assert authority(root, registry, session, turn)["targets"] == [
+        {"goal_id": "other", "agent_id": "peer"}
+    ]
+    with pytest.raises(ValueError, match="not authorized"):
+        deliver(root, registry, session=session, turn=turn, request=request)
+    assert len(pending(root, "research", "worker")["items"]) == 1
+
+    data["goals"][0]["activation_state"] = "active"
+    registry.write_text(json.dumps(data))
+    assert deliver(root, registry, session=session, turn=turn, request=request) == {
+        **first, "replayed": True
+    }
+
+
+def test_stopped_or_invalid_goal_is_excluded_from_lark_and_goal_chat(fixture):
+    root, registry, session, turn, request = fixture
+    data = json.loads(registry.read_text())
+    data["goals"][0]["activation"] = {
+        "schema_version": "loopx_goal_activation_v1", "state": "stopped"
+    }
+    registry.write_text(json.dumps(data))
+
+    goal_session = {**session, "channel_id": "goal.research", "goal_id": "research"}
+    assert authority(root, registry, goal_session, turn)["targets"] == []
+    with pytest.raises(ValueError, match="not authorized"):
+        deliver(root, registry, session=goal_session, turn=turn, request=request)
+
+    lark_session = {**session, "channel_id": "manager.external.group"}
+    lark_turn = {**turn, "origin": "lark"}
+    _write(_root(root) / "policy.json", {
+        "schema_version": POLICY_SCHEMA,
+        "sources": {lark_session["channel_id"]: {
+            "sender_ids": ["owner"], "targets": [request]
+        }},
+    })
+    register_ingress(root, session_id=session["session_id"],
+                     client_turn_id=turn["client_turn_id"],
+                     channel=lark_session["channel_id"], sender_id="owner",
+                     message=turn["message"], source_id="lark:original")
+    assert authority(root, registry, lark_session, lark_turn)["targets"] == []
+    with pytest.raises(ValueError, match="not authorized"):
+        deliver(root, registry, session=lark_session, turn=lark_turn, request=request)
+
+    data["goals"][0]["activation"]["state"] = "unreadable"
+    registry.write_text(json.dumps(data))
+    assert authority(root, registry, session, turn)["targets"] == [
+        {"goal_id": "other", "agent_id": "peer"}
+    ]
+
+
 @pytest.mark.parametrize("changes", [
     {"channel_id": "goal.other"}, {"goal_id": "other"}, {"goal_id": ""},
 ])