fix(chat): retain steward identity across executor changes

loopx/control_plane/collaboration/conversation_binding.ts

@@ -156,8 +156,10 @@ function binding(value: unknown): JsonObject {
       throw new EffectRuntimeRequestError("ambiguous Agent target grant");
     }
   }
-  if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group || row.executor_endpoint_id !== "codex")) {
-    throw new EffectRuntimeRequestError("managed identity requires a private native steward");
+  // The prepared identity survives executor changes. Native execution still
+  // requires the Codex binding at stewardManagedBasis, not at storage read.
+  if (row.managed_actor !== undefined && (row.context_kind !== "steward" || group)) {
+    throw new EffectRuntimeRequestError("managed identity requires a private steward");
   }
   return {
     schema_version: BINDING_SCHEMA, binding_id: ref(row.binding_id, "binding identity"),

tests/control_plane_ts/conversation_binding.test.ts

@@ -4,6 +4,7 @@ import {planConversationBinding, resolveBoundConversation, planBoundConversation
   stewardCommand, authorizeStewardCreation, resolveConversationAgentTarget,
   listLocalBoundSessions} from "../../loopx/control_plane/collaboration/conversation_binding.ts";
 import {resolveConversationScope, projectConversationIdentity} from "../../loopx/control_plane/collaboration/conversation_scope.ts";
+import {stewardManagedBasis} from "../../loopx/control_plane/collaboration/steward_managed_mode.ts";
 
 const project = {kind: "project_workspace", project_ref: "a".repeat(24), workspace_path: "/authorized/notes",
   audience: "local_owner", grant: "workspace_read"};
@@ -17,6 +18,38 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision
 const request = {current, expected_revision: 0, operation: "configure", binding: row, observation,
   available_projects: [project]};
 
+for (const endpoint of ["codex", "antigravity", "kiro-cli"]) {
+  test(`retained steward identity permits bound reads after selecting ${endpoint}`, () => {
+    const actor = {schema_version: "loopx_steward_managed_actor_v0",
+      goal_id: `steward-agent-${"8".repeat(24)}`, goal_instance_id: null, agent_id: "steward",
+      session_id: "7".repeat(32), source_ref: "6".repeat(24), creation_operation_id: "original-creation"};
+    const steward = {...row, binding_id: "8".repeat(24), transport_ref: "owner-steward",
+      provider_ref: "9".repeat(24), context_kind: "steward", grant: "portfolio_read", goal_ids: [],
+      executor_endpoint_id: endpoint, managed_actor: actor};
+    const configured = {...planConversationBinding(request).state, bindings: [row, steward]};
+    const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};
+    const notes = resolveBoundConversation({current: configured, binding_id: row.binding_id,
+      observation, available_projects: [project], ...source});
+    assert.deepEqual(notes.binding, row, "An unrelated steward cannot block project admission");
+    const selected = resolveBoundConversation({current: configured, binding_id: steward.binding_id,
+      observation: {...observation, transport_ref: steward.transport_ref, provider_ref: steward.provider_ref},
+      available_projects: [project], ...source});
+    assert.deepEqual(selected.binding.managed_actor, actor);
+    assert.equal(selected.binding.executor_endpoint_id, endpoint);
+    for (const invalid of [{...steward, context_kind: "project", grant: "workspace_read"},
+      {...steward, audience: "group", group_refs: ["5".repeat(24)]},
+      {...steward, managed_actor: {...actor, agent_id: "another-agent"}}]) {
+      assert.throws(() => resolveBoundConversation({current: {...configured, bindings: [row, invalid]},
+        binding_id: row.binding_id, observation, available_projects: [project], ...source}));
+    }
+    if (endpoint !== "codex") {
+      assert.throws(() => stewardManagedBasis({binding: steward,
+        session: {steward_context: selected.context}, observation}), /original idle private steward/,
+      "Retaining an identity cannot authorize unsupported native execution");
+    }
+  });
+}
+
 test("unfinished completed work requires typed intent and latest settled native/source proof, once", () => {
   const configured = planConversationBinding(request).state;
   const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};

tests/test_native_steward_managed_mode.py

@@ -74,6 +74,59 @@ def test_native_preparation_preserves_original_and_creates_one_real_actor(stewar
     assert transport.bindings.read()["bindings"][-1]["managed_actor"] == actor
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_retained_actor_on_another_executor_does_not_block_captured_project_work(steward, tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    store, runtime, external, transport, binding, capture, _ = steward
+    sid, service = original(steward)
+    storage(runtime, provider)
+    preview = service.prepare(binding["binding_id"], session_id=sid)
+    actor = service.prepare(binding["binding_id"], session_id=sid,
+        expected_plan_id=preview["plan"]["plan_id"], execute=True)["actor"]
+    event = external.event("notes-app", "captured-project-work", "Explain this project")
+    admit = transport.core.admit
+
+    def unavailable(**_kwargs):
+        raise OSError("source captured before Core admission")
+
+    monkeypatch.setattr(transport.core, "admit", unavailable)
+    with pytest.raises(OSError, match="before Core admission"):
+        transport.admit("notes-app", event)
+    monkeypatch.setattr(transport.core, "admit", admit)
+    path = next(path for path in transport.root.glob("*.json")
+                if json.loads(path.read_text())["event"]["message_id"] == event["message_id"])
+    assert json.loads(path.read_text())["status"] == "captured"
+    # A persisted executor selection retains the previously prepared identity.
+    # This compatibility fixture is isolated from all real bindings and Goals.
+    current = transport.bindings.read()
+    selected = next(row for row in current["bindings"] if row["binding_id"] == binding["binding_id"])
+    selected["executor_endpoint_id"] = "antigravity"
+    transport.bindings.path.write_text(json.dumps(current))
+    before = capture.read_bytes()
+    transport.reconcile_request(path)
+    row = transport.core.read_request(json.loads(path.read_text())["request_ref"])
+    assert row["status"] == "accepted"
+    result = finish(runtime, row)
+    assert result["status"] == "completed"
+    transport.reconcile()
+    delivered = json.loads(path.read_text())
+    assert delivered["status"] == "delivered" and delivered["deliveries"]["terminal"]["verified"]
+    reply = external.messages[delivered["deliveries"]["terminal"]["attempt"]["message_ref"]]
+    assert reply["chat_id"] == event["chat_id"]
+    assert [message["text"] for message in store.messages(row["session_id"]) if message["role"] == "user"] == [event["content"]]
+    packets = [json.loads(line) for line in capture.read_bytes()[len(before):].splitlines()]
+    assert sum(packet["method"] == "turn/start" for packet in packets) == 1
+    assert transport.reconcile() == 0
+    transport.ingress.observe()
+    assert transport.ingress.health(binding["binding_id"])["status"] == "verified"
+    assert next(item for item in transport.bindings.read()["bindings"]
+                if item["binding_id"] == binding["binding_id"])["managed_actor"] == actor
+    before_prepare = capture.read_bytes(), runtime.registry_path.read_bytes()
+    with pytest.raises(ValueError):
+        service.prepare(binding["binding_id"], session_id=sid)
+    assert (capture.read_bytes(), runtime.registry_path.read_bytes()) == before_prepare
+
+
 @pytest.mark.parametrize("fault", ["queued", "provider", "stale_preview", "foreign_session"])
 def test_rejected_preparation_has_no_native_goal_or_model_effect(steward, fault):
     store, runtime, provider, _, binding, capture, _ = steward