fix(chat): preserve project identity in source scope resolution

loopx/control_plane/collaboration/__init__.py

@@ -17,9 +17,11 @@ def conversation_trigger(mode: str | None = None, **evidence: bool) -> dict[str,
 
 
 def conversation_scope(session: dict[str, Any], *, origin: str | None = None) -> dict[str, Any]:
+    # Core distinguishes an absent legacy Goal field from an explicit project null.
     return effect_runtime_result("collaboration.conversation.scope", {
-        "channel_id": session.get("channel_id"), "goal_id": session.get("goal_id"),
-        "project_context": session.get("project_context"), "steward_context": session.get("steward_context"),
+        **{key: session[key] for key in (
+            "channel_id", "goal_id", "project_context", "steward_context",
+        ) if key in session},
         **({"origin": origin} if origin is not None else {}),
     })
 

loopx/control_plane/collaboration/conversation_scope.ts

@@ -127,6 +127,9 @@ export function resolveConversationScope(input: Record<string, unknown>): Conver
         return {kind: "project_workspace", goal_ids: [], private_conversation: false};
       }
     } catch { /* Incomplete host identity grants no context. */ }
+    // Explicit null identifies a project Session; a missing legacy Goal field
+    // is different and must not be manufactured by an observation adapter.
+    return {kind: "unavailable", goal_ids: [], private_conversation: false};
   }
   if (goal === "loopx-manager" && (input.origin === undefined || input.origin === "lark" || input.origin === "external" || input.origin === "web")) {
     try {
@@ -136,6 +139,11 @@ export function resolveConversationScope(input: Record<string, unknown>): Conver
       }
     } catch { /* An incomplete steward identity cannot authorize a portfolio. */ }
   }
+  // Declared context must match a typed identity above; a legacy channel cannot
+  // turn inconsistent recovered project metadata into broader authority.
+  if (input.project_context != null) {
+    return {kind: "unavailable", goal_ids: [], private_conversation: false};
+  }
   if (channel === "manager") {
     return {kind: "owner_portfolio", goal_ids: null, private_conversation: true};
   }

loopx/control_plane/collaboration/source_grant_observation.py

@@ -94,9 +94,9 @@ def _source_context_grant(
     return cast(dict[str, Any], effect_runtime_result(
         "collaboration.source.context_authority",
         {
-            "conversation": {key: session.get(key) for key in (
+            "conversation": {key: session[key] for key in (
                 "session_id", "goal_id", "channel_id", "project_context", "steward_context",
-            )} | {"origin": turn.get("origin", "unknown")},
+            ) if key in session} | {"origin": turn.get("origin", "unknown")},
             "turn": {"client_turn_id": turn.get("client_turn_id"), "origin": turn.get("origin"),
                      "message_digest": _hash(turn.get("message"))} if ingress is not None else None,
             "ingress": {key: ingress[key] for key in (

tests/control_plane_ts/conversation_scope.test.ts

@@ -30,3 +30,32 @@ test("a project name, role label or external anchor cannot grant a wider scope",
   assert.deepEqual(resolveConversationScope({channel_id: "manager.external.audience", goal_id: "alpha"}),
     {kind: "external_audience", goal_ids: [], private_conversation: false});
 });
+
+test("declared project identity cannot fall back to a legacy manager channel", () => {
+  const project = {kind: "project_workspace", project_ref: "a".repeat(24),
+    workspace_path: "/synthetic/project", audience: "local_owner", grant: "workspace_read"};
+  const bound = {...project, audience: "bound_owner", binding_id: "b".repeat(24),
+    source_ref: "c".repeat(24), provider_ref: "d".repeat(24), operator_ref: "e".repeat(24)};
+  const group = {...bound, audience: "bound_group", filesystem_scope: "workspace_only",
+    group_ref: "f".repeat(24), topic_ref: "0".repeat(24)};
+  for (const context of [project, bound, group, {...project, grant: "invalid"}]) {
+    for (const identity of [
+      {channel_id: "manager", goal_id: null, origin: "web"},
+      {channel_id: "manager", goal_id: "alpha", origin: "web"},
+      {channel_id: "manager", goal_id: "loopx-manager", origin: "web"},
+      {channel_id: "manager.external.audience", goal_id: null, origin: "lark"},
+    ]) {
+      assert.deepEqual(resolveConversationScope({...identity, project_context: context}),
+        {kind: "unavailable", goal_ids: [], private_conversation: false});
+    }
+  }
+  // A nullable transport projection alone does not change a legacy manager.
+  for (const goal of [undefined, "alpha", "loopx-manager"]) {
+    assert.deepEqual(resolveConversationScope({channel_id: "manager", goal_id: goal, project_context: null}),
+      {kind: "owner_portfolio", goal_ids: null, private_conversation: true});
+  }
+  for (const channel of ["manager", "manager.external.audience"]) {
+    assert.deepEqual(resolveConversationScope({channel_id: channel, goal_id: null}),
+      {kind: "unavailable", goal_ids: [], private_conversation: false});
+  }
+});

tests/test_manager_context_roundtrip.py

@@ -11,6 +11,7 @@
     _root,
     _write,
     acknowledge,
+    authority,
     deliver,
     pending,
     register_ingress,
@@ -27,6 +28,7 @@
     reply_status,
 )
 from loopx.chat_store import ChatSessionStore
+from loopx.control_plane.collaboration import conversation_scope
 
 
 @pytest.fixture
@@ -158,6 +160,59 @@ def test_owner_return_does_not_create_source_policy(flow, project):
     assert not list(_root(root).glob("policy*"))
 
 
+@pytest.mark.parametrize("audience", ["local_owner", "bound_owner"])
+@pytest.mark.parametrize("drift", ["channel", "goal", "context", "context_missing", "context_null"])
+def test_recovered_project_identity_cannot_publish_as_a_manager(flow, audience, drift):
+    root, registry, _, _ = flow
+    store = ChatSessionStore(root / "separate-chat-host")
+    project = {"kind": "project_workspace", "project_ref": "a" * 24,
+               "workspace_path": str(root), "audience": audience, "grant": "workspace_read"}
+    channel = "project." + project["project_ref"]
+    if audience == "bound_owner":
+        project.update(binding_id="b" * 24, source_ref="c" * 24,
+                       provider_ref="d" * 24, operator_ref="e" * 24)
+        channel = "project.external." + project["binding_id"] + "." + project["source_ref"]
+    session = store.create_session(goal_id=None, agent_id="codex", adapter_kind="codex_app_server",
+                                   upstream_thread_id="original", channel_id=channel, project_context=project)
+    turn, _ = store.create_turn(session["session_id"], client_turn_id="question",
+                                message="Read-only project question", origin="web")
+    # Inject inconsistent recovered metadata only in this disposable store.
+    # The registered receiver exists; the project never gained its source grant.
+    path = store.root / "sessions" / session["session_id"] / "session.json"
+    saved = json.loads(path.read_text())
+    saved["channel_id"] = "manager"
+    if drift == "goal":
+        saved["goal_id"] = "research"
+    elif drift == "context":
+        saved["project_context"]["grant"] = "invalid"
+    elif drift == "context_missing":
+        saved.pop("project_context")
+    elif drift == "context_null":
+        saved["project_context"] = None
+    path.write_text(json.dumps(saved))
+    observed = store.load_session(session["session_id"])
+    before = {p.relative_to(_root(root)): p.read_bytes() for p in _root(root).rglob("*.json")}
+    with pytest.raises(ValueError, match="context recipient is not authorized or registered"):
+        deliver(root, registry, session=observed, turn=turn,
+                request={"goal_id": "research", "agent_id": "worker"}, source_store=store)
+    assert {p.relative_to(_root(root)): p.read_bytes() for p in _root(root).rglob("*.json")} == before
+    assert not pending(root, "research", "worker")["items"]
+    assert json.loads(path.read_text()) == saved
+    assert store.load_turn(session["session_id"], turn["turn_id"]) == turn
+
+
+def test_legacy_manager_source_preserves_missing_goal_instead_of_project_null(flow):
+    root, registry, store, _ = flow
+    session = store.create_session(goal_id="loopx-manager", agent_id="codex", adapter_kind="codex_app_server",
+                                   upstream_thread_id="original", channel_id="manager")
+    turn, _ = store.create_turn(session["session_id"], client_turn_id="question", message="Original question", origin="web")
+    legacy = {key: value for key, value in session.items() if key != "goal_id"}
+    assert conversation_scope(legacy)["kind"] == "owner_portfolio"
+    assert conversation_scope({**legacy, "goal_id": None})["kind"] == "unavailable"
+    assert authority(root, registry, legacy, turn)["targets"] == [{"goal_id": "research", "agent_id": "worker"}]
+    assert authority(root, registry, {**legacy, "goal_id": None}, turn) == {"mode": "unavailable", "targets": []}
+
+
 @pytest.mark.parametrize("explicit_source", [False, True], ids=["legacy-root", "pinned-root"])
 def test_other_chat_host_cannot_delay_original_host_return(flow, explicit_source):
     root, registry, store, create = flow