refactor(operations): keep source resolution inside managed admission

Signed-off-by: huangruiteng <huangrt01@163.com>

Author: huangruiteng

docs/architecture/rfcs/human-confirmed-domain-operations-v0.md

@@ -470,7 +470,8 @@ option, `--codex-operation-source-route-json
 The selector is registered-audience routing, not authentication, a session
 replacement or an execution permit; the model cannot retarget it. Managed duplicate
 bindings collapse. No registered audience retains the historical null route;
-non-managed adapters retain their previous count-based implicit routing behavior. Store
+non-managed adapters retain their previous no-source-route projection and do not
+invoke this managed resolver. Store
 preparation freezes the selected audience in the original confirmation digest.
 
 Source-audience selection and the preparation prompt form a bounded backend

docs/architecture/rfcs/human-confirmed-domain-operations-v0.zh-CN.md

@@ -365,8 +365,8 @@ Core PR 仍须 owner review，不在合并前自行安装。
 受众时，必须由宿主显式传入 `--codex-operation-source-route-json
 '{"host_surface":"codex-app","thread_id":"REGISTERED_THREAD"}'`。
 这只是已登记的回传受众，不是认证、session 替换或执行许可；模型不能改投。
-受管重复绑定会去重；无登记受众保持历史 null 路由，非受管 adapter 保留既有计数式隐式
-路由行为。准备落盘后，返回受众进入原确认摘要，不可修改。
+受管重复绑定会去重；无登记受众保持历史 null 路由，非受管 adapter 保留既有
+无 source-route 投影，不调用这条受管解析。准备落盘后，返回受众进入原确认摘要，不可修改。
 
 源受众选择与准备提示构成有界后台切片，不是完整产品交付。下面的个人工作台
 配套改动仍在同一计划内单独完成首屏评审后交付。工作台在可见时读取规范

loopx/chat_action_normalization.py

@@ -237,15 +237,16 @@ def _normalize(
             managed_source = executor.get("kind") == "managed_turn"
             if not managed_source and "source_route" in values:
                 raise ValueError("source route selection requires a managed executor")
-            from .control_plane.effect_runtime import effect_runtime_result
+            source_route = None
+            if managed_source:
+                from .control_plane.effect_runtime import effect_runtime_result
 
-            source_route = effect_runtime_result(
-                "operation.source_route.resolve",
-                {"goal_id": goal_id, "agent_id": agent_id,
-                 "bindings": (goal.get("coordination") or {}).get("thread_agent_bindings", []),
-                 "selected_route": values.get("source_route"),
-                 "ambiguity_policy": "reject" if managed_source else "legacy_null"},
-            )["source_route"]
+                source_route = effect_runtime_result(
+                    "operation.source_route.resolve",
+                    {"goal_id": goal_id, "agent_id": agent_id,
+                     "bindings": (goal.get("coordination") or {}).get("thread_agent_bindings", []),
+                     "selected_route": values.get("source_route")},
+                )["source_route"]
             expires_at = parse_timestamp(
                 _text(values.get("expires_at"), field="expires_at", limit=80)
             )

loopx/control_plane/work_items/operation_agent_handoff.ts

@@ -34,14 +34,12 @@ export function resolveOperationSourceRoute(input: JsonObject): JsonObject {
   const agent = id(input.agent_id, "agent_id");
   const bindings = Array.isArray(input.bindings) ? input.bindings : [];
   const routes = new Map<string, JsonObject>();
-  let registeredCount = 0;
   for (const raw of bindings) {
     if (raw === null || typeof raw !== "object" || Array.isArray(raw)) continue;
     const route = raw as JsonObject;
     if (route.agent_id !== agent || typeof route.host_surface !== "string"
       || typeof route.thread_id !== "string" || !ID.test(route.host_surface) || !ID.test(route.thread_id)) continue;
     const audience = {goal_id: goal, agent_id: agent, host_surface: route.host_surface, thread_id: route.thread_id};
-    registeredCount += 1;
     routes.set(JSON.stringify(audience), audience);
   }
   if (input.selected_route != null) {
@@ -55,11 +53,6 @@ export function resolveOperationSourceRoute(input: JsonObject): JsonObject {
     requireThat(routes.has(JSON.stringify(audience)), "operation source route is not registered for this Goal and Agent");
     return {source_route: audience};
   }
-  if (input.ambiguity_policy === "legacy_null") {
-    // Preserve the old non-managed count-based behavior, including duplicated
-    // registry rows. Distinct-audience de-duplication is managed opt-in only.
-    return {source_route: registeredCount === 1 ? routes.values().next().value : null};
-  }
   if (routes.size > 1) throw new EffectRuntimeRequestError(
     "operation source route is ambiguous; select a registered return audience in the host invocation",
     "operation_source_route_ambiguous");

tests/control_plane_ts/operation_agent_handoff.test.ts

@@ -12,11 +12,6 @@ test("registered return audience is explicit under ambiguity and never executor
   assert.equal(resolveOperationSourceRoute({...value, bindings: []}).source_route, null);
   const multiple = {...value, bindings: [route, {...route, thread_id: "historical"}]};
   assert.throws(() => resolveOperationSourceRoute(multiple), {code: "operation_source_route_ambiguous"});
-  // Unchanged non-managed callers retain their historical null-on-ambiguity
-  // routing behavior; the normalizer never lets a model choose this policy.
-  assert.equal(resolveOperationSourceRoute({...multiple, ambiguity_policy: "legacy_null"}).source_route, null);
-  assert.equal(resolveOperationSourceRoute({...value, bindings: [route, route], ambiguity_policy: "legacy_null"}).source_route, null);
-  assert.deepEqual(resolveOperationSourceRoute({...value, ambiguity_policy: "legacy_null"}), resolveOperationSourceRoute(value));
   const selector = {host_surface: route.host_surface, thread_id: route.thread_id};
   assert.deepEqual(resolveOperationSourceRoute({...multiple, selected_route: selector}), resolveOperationSourceRoute(value));
   for (const selected_route of ["original", {...selector, thread_id: "unregistered"},

tests/test_chat_operation_actions.py

@@ -192,6 +192,23 @@ def test_non_managed_prepare_rejects_source_selector_even_when_null(tmp_path: Pa
     assert store.list() == []
 
 
+def test_non_managed_prepare_does_not_mount_managed_source_resolution(tmp_path: Path, monkeypatch) -> None:
+    from loopx.control_plane import effect_runtime
+
+    original = effect_runtime.effect_runtime_result
+    calls = []
+
+    def record(method, *args, **kwargs):
+        calls.append(method)
+        return original(method, *args, **kwargs)
+
+    monkeypatch.setattr(effect_runtime, "effect_runtime_result", record)
+    service, store = _service(tmp_path)
+    proposal = service.preview(_request())
+    assert "source_route" not in store.load(proposal["proposal_id"])["normalized_parameters"]
+    assert "operation.source_route.resolve" not in calls
+
+
 def test_managed_prepare_selects_registered_return_audience_without_rebinding_executor(tmp_path: Path) -> None:
     from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler