Recover already admitted Steward continuations after a lost acknowledgement

docs/product/releases/v2.3.0.md

@@ -62,6 +62,7 @@ loopx doctor
 - Research shows the captured Turn range and available usage evidence, provides a readable report, and preserves a bookmarked result while part of the source is unavailable. Home can open an existing report edition. [#7882](https://github.com/loopx-project/loopx/pull/7882), [#7973](https://github.com/loopx-project/loopx/pull/7973), [#8017](https://github.com/loopx-project/loopx/pull/8017), [#8116](https://github.com/loopx-project/loopx/pull/8116), [#8146](https://github.com/loopx-project/loopx/pull/8146), [#8226](https://github.com/loopx-project/loopx/pull/8226)
 - Conversations can steer a supported active Turn and edit or cancel pending messages. Recovered native replies keep the request and answer together. Use the original conversation to review the result. [#8059](https://github.com/loopx-project/loopx/pull/8059), [#8096](https://github.com/loopx-project/loopx/pull/8096), [#8121](https://github.com/loopx-project/loopx/pull/8121), [#8144](https://github.com/loopx-project/loopx/pull/8144), [#8269](https://github.com/loopx-project/loopx/pull/8269)
 - After a service exit, the Steward can resume its saved Codex working thread when the account, settings, and source binding still match. [#8379](https://github.com/loopx-project/loopx/pull/8379)
+- If acknowledgement of an admitted Steward continuation is lost, the original conversation can recover that same Turn and return its result once.
 - Issue-Fix verifies correction events and can return an eligible result to its original request. Failed Todos can produce owner attention without assigning a new worker. Anti-Entropy is available for tasks that explicitly declare a retirement basis. [#7860](https://github.com/loopx-project/loopx/pull/7860), [#7861](https://github.com/loopx-project/loopx/pull/7861), [#8129](https://github.com/loopx-project/loopx/pull/8129)
 
 ## Quality & Testing
@@ -207,7 +208,7 @@ loopx slash-commands --uninstall --surface agy
 
 **Activation:** Install the optional account package into the LoopX Python environment and restart its serving owner. In Settings → Model Provider configuration, select Steward API use → Prefer API with an existing authorized API key.
 
-**Validation:** Reload the selector and check its saved value. Read the working-Agent policy separately.
+**Validation:** Reload the selector and check its saved value. Run the native account extension doctor and inspect `consumer_readiness` for the worker and the configured Steward separately. Both successful checks must identify the same relay instance. [#8394](https://github.com/loopx-project/loopx/pull/8394)
 
 **Disable / rollback:** Select Disabled for the Steward's API role. The existing worker fallback setting controls working Agents separately.
 
@@ -356,7 +357,7 @@ python -m benchmark.edgebench.run --help
 
 **Activation:** Opt in per source-pack registration with `--case-study-json`. Inspect existing packs first. Register the case's pack under its own benchmark child Goal; a global switch or model execution is not needed.
 
-**Validation:** List the registered packs, then open the case in Research. Follow a cited line back to its exact saved source. Changed sources remain historical, and unavailable translations remain explicit.
+**Validation:** List the registered packs, then open the case in Research. Check that a narrow chart still shows the saved observations in order, including their complete labels and values. A missing value stays distinct from a measured zero. Follow a cited line back to its exact saved source. Changed sources remain historical, and unavailable translations remain explicit. [#8387](https://github.com/loopx-project/loopx/pull/8387)
 
 **Disable / rollback:** Return to Research records and analysis or select a previous pack. Omit the companion on future registrations. Editing a companion creates a new version; retain existing source and reading versions.
 
@@ -496,6 +497,7 @@ loopx --format json goal-acceptance inspect --goal-id "$goal_id"
 - 从研究记录查看已捕获的 Turn 范围、用量与可读报告，补读期间保留书签结果。首页可以回看报告版本。 [#7882](https://github.com/loopx-project/loopx/pull/7882), [#7973](https://github.com/loopx-project/loopx/pull/7973), [#8017](https://github.com/loopx-project/loopx/pull/8017), [#8116](https://github.com/loopx-project/loopx/pull/8116), [#8146](https://github.com/loopx-project/loopx/pull/8146), [#8226](https://github.com/loopx-project/loopx/pull/8226)
 - 对话支持修改或撤销待发送项，并可在受支持的会话中引导当前 Turn。恢复后仍从原请求读回答案。 [#8059](https://github.com/loopx-project/loopx/pull/8059), [#8096](https://github.com/loopx-project/loopx/pull/8096), [#8121](https://github.com/loopx-project/loopx/pull/8121), [#8144](https://github.com/loopx-project/loopx/pull/8144), [#8269](https://github.com/loopx-project/loopx/pull/8269)
 - 服务退出后，若账号、设置及来源绑定仍一致，管家可续用已保存的 Codex 工作线程。 [#8379](https://github.com/loopx-project/loopx/pull/8379)
+- 管家续作已入队但确认丢失时，原会话可找回同一个 Turn，并回送一次结果。
 - Issue-Fix 可验证修订事件并回送原请求。失败 Todo 能提醒负责人。显式声明退役对象的任务可以用 Anti-Entropy 检查源码。 [#7860](https://github.com/loopx-project/loopx/pull/7860), [#7861](https://github.com/loopx-project/loopx/pull/7861), [#8129](https://github.com/loopx-project/loopx/pull/8129)
 
 ### 质量与测试
@@ -642,7 +644,7 @@ loopx slash-commands --uninstall --surface agy
 
 **启用：** Codex 账号 API 优先级。在 LoopX 的 Python 环境安装可选账号扩展并重启服务，在「设置 → 模型提供方」用已授权 key 选择「管家 API → API 优先」。
 
-**验证：** 刷新读回选择项，另行检查执行 Agent 的订阅优先策略。
+**验证：** 刷新读回选择项。运行原生账号扩展的 Doctor，在 `consumer_readiness` 分别核对执行 Agent 和已配置的管家通道。两项成功读回应指向同一 relay 实例。 [#8394](https://github.com/loopx-project/loopx/pull/8394)
 
 **停用 / 回退：** 将管家 API 角色设为「禁用」。执行 Agent 按其独立 fallback 设置管理。
 
@@ -791,7 +793,7 @@ python -m benchmark.edgebench.run --help
 
 **启用：** 每次注册 source pack 时用 `--case-study-json` 显式附加阅读稿，先列出已有 pack。将该案例的 pack 注册到对应 benchmark child Goal，无需全局开关或模型执行。
 
-**验证：** 列出已注册 pack，在 Research 打开案例，沿引用行回到原始来源。来源变化时保留历史提示，缺少翻译时明确显示原文。
+**验证：** 列出已注册 pack，在 Research 打开案例。在窄屏图表中核对观察顺序、完整标签和数值，缺失值与测得的零应分别显示。沿引用行回到原始来源。来源变化时保留历史提示，缺少翻译时明确显示原文。 [#8387](https://github.com/loopx-project/loopx/pull/8387)
 
 **停用 / 回退：** 返回 Research records and analysis 或选择原 pack。后续注册省略阅读稿；修改阅读稿时创建新版本，保留既有来源与阅读版本。
 

loopx/capabilities/native_chat/external_conversations.py

@@ -157,10 +157,9 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
         elif selected_binding is None:
             selected_binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"]
         reservation = row.get("context_recovery") or {}
-        admitted = (store.turn_for_client(session["session_id"], reservation["client_turn_id"])
-            if (reservation.get("routing_attempt") or reservation.get("quota_attempt") or reservation.get("preparation_attempt")
-                or reservation.get("read_only_review") or reservation.get("work_attempt")
-                or reservation.get("metadata_review"))
+        client_turn_id = reservation.get("client_turn_id")
+        admitted = (store.turn_for_client(session["session_id"], client_turn_id)
+            if isinstance(client_turn_id, str) and client_turn_id
             and reservation.get("failed_turn_id") == turn["turn_id"]
             and (retained_review is None or reservation.get("work_attempt") == 1
                  or reservation.get("legacy_result_review")) else None)

loopx/control_plane/collaboration/conversation_binding.ts

@@ -851,20 +851,24 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
       || typeof failed.upstream_turn_id !== "string" || !failed.upstream_turn_id) return no;
   const retryKind = failed.error_code === "usage_limit_exceeded" ? "quota" : "routing";
   const attemptKey = `${retryKind}_attempt`;
+  const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}`;
+  const attempt = previous?.[attemptKey];
+  const reservedClient = Number.isSafeInteger(attempt) && Number(attempt) >= 1 && Number(attempt) <= 3
+    ? `${client}-${retryKind}-${attempt}`
+    : attempt === undefined && steward && previous?.client_turn_id === client ? client : null;
   if (recovery.admitted_turn != null && previous !== null && previous.failed_turn_id === failed.turn_id
-      && Number.isSafeInteger(previous[attemptKey]) && Number(previous[attemptKey]) >= 1
-      && Number(previous[attemptKey]) <= 3 && previous.version === version) {
+      && previous.version === version && reservedClient !== null) {
     // A crash after queue admission must adopt that exact canonical Turn,
     // even if it has already executed. This branch never creates model work.
     const admitted = requireJsonObject(recovery.admitted_turn, "admitted recovery Turn");
-    const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}-${retryKind}-${previous[attemptKey]}`;
-    if (previous.client_turn_id !== client || admitted.client_turn_id !== client
+    const origin = reservedClient === client && row.origin === undefined ? "lark" : row.origin;
+    if (previous.client_turn_id !== reservedClient || admitted.client_turn_id !== reservedClient
         || admitted.session_id !== current.session_id || admitted.turn_id === failed.turn_id
-        || typeof row.message !== "string" || typeof row.origin !== "string"
-        || admitted.message !== row.message || admitted.origin !== row.origin
+        || typeof row.message !== "string" || typeof origin !== "string"
+        || admitted.message !== row.message || admitted.origin !== origin
         || !Array.isArray(admitted.attachments ?? []) || (admitted.attachments as unknown[] | undefined)?.length) return no;
     return {operation: "admit_turn", recovery_candidate: true,
-      recovery_eligible: recovery.source_verified === true, client_turn_id: client,
+      recovery_eligible: recovery.source_verified === true, client_turn_id: reservedClient,
       admitted_turn_id: sessionIdentity(admitted.turn_id, "admitted recovery Turn"),
       session_id: current.session_id, native_thread_ids: []};
   }
@@ -874,7 +878,6 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
   const routing = planRoutingRecovery(recovery, row, current, failed, previous, version);
   if (routing !== null) return routing;
   if (!steward) return no;
-  const client = `external-${ref(row.request_ref, "original request")}-recovery-${version}`;
   if (previous && (previous.version !== version || previous.failed_turn_id !== row.turn_id
       || previous.client_turn_id !== client)) return no;
   const history = current.upstream_context_history ?? [];

tests/control_plane_ts/conversation_binding.test.ts

@@ -1274,6 +1274,40 @@ test("the App steward is an explicit source preference, never a new grant or pee
     app_steward_entry: true, observation, available_projects: [project], sessions: []}), /selected original steward/);
 });
 
+test("a legacy Steward recovery adopts its exact admitted Turn after the session advances", () => {
+  const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: []};
+  const configured = planConversationBinding({...request, binding: selected}).state;
+  const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};
+  const context = resolveBoundConversation({current: configured, binding_id: row.binding_id,
+    observation, available_projects: [project], ...source}).context;
+  const ref = "f".repeat(24), message = "Original synthetic request";
+  const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "codex",
+    upstream_mode: "chat", status: "ready", upstream_thread_id: "updated-native", active_turn_id: null,
+    manager_context_version: 22, steward_context: context,
+    channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`};
+  const failed = {session_id: "original", turn_id: "failed", status: "failed", delta_count: 0,
+    response: null, error_code: "context_window_exceeded", upstream_turn_id: "native-failed"};
+  const reservation = {version: 22, failed_turn_id: "failed", client_turn_id: `external-${ref}-recovery-22`};
+  const admitted = {session_id: "original", turn_id: "continued", status: "completed",
+    client_turn_id: reservation.client_turn_id, origin: "lark", message, attachments: []};
+  const packet = {binding: selected, current_session: session, request: {request_ref: ref, command: null,
+    status: "accepted", delivery_verified: true, binding_id: row.binding_id, session_id: "original",
+    turn_id: "failed", source, message, origin: "lark", context_recovery: reservation},
+  context_recovery: {version: 22, failed_turn: failed, steering_observed: false, source_verified: true,
+    native_observation: null, admitted_turn: admitted}};
+  const plan = planBoundConversationRequest(packet);
+  assert.equal(plan.recovery_eligible, true);
+  assert.equal(plan.admitted_turn_id, "continued");
+  assert.deepEqual(plan.native_thread_ids, []);
+  assert.equal(planBoundConversationRequest({...packet, context_recovery: {
+    ...packet.context_recovery, source_verified: false}}).recovery_eligible, false);
+  for (const change of [{session_id: "foreign"}, {client_turn_id: "foreign"}, {turn_id: failed.turn_id},
+    {message: "Changed request"}, {origin: "external"}, {attachments: [{type: "image"}]}]) {
+    assert.equal(planBoundConversationRequest({...packet, context_recovery: {
+      ...packet.context_recovery, admitted_turn: {...admitted, ...change}}}).recovery_eligible, false);
+  }
+});
+
 for (const kind of ["project", "steward"]) for (const retryKind of ["routing", "quota"]) {
   test(`${kind} ${retryKind} recovery is bounded, source-bound and independent of context migrations`, () => {
     const selected = kind === "steward" ? {...row, context_kind: kind, grant: "portfolio_read", goal_ids: []} : row;

tests/test_native_steward_private.py

@@ -247,14 +247,23 @@ def test_existing_scanner_automatically_continues_legacy_failure_and_returns_ori
         store.update_session(row["session_id"], manager_context_version=MANAGER_CONTEXT_VERSION)
     if fault in {"reservation-crash", "post-enqueue-crash"}:
         enqueue = runtime.enqueue_turn
+        continued = None
         def crash(**kwargs):
+            nonlocal continued
             if fault == "post-enqueue-crash":
-                enqueue(**kwargs)
+                continued, _ = enqueue(**kwargs)
             raise OSError("simulated admission crash")
         monkeypatch.setattr(runtime, "enqueue_turn", crash)
         with pytest.raises(OSError, match="admission"):
             transport.reconcile()
         monkeypatch.setattr(runtime, "enqueue_turn", enqueue)
+        if continued is not None:
+            assert finish(runtime, continued)["status"] == "completed"
+            assert transport.core.read_request(row["request_ref"])["turn_id"] == row["turn_id"]
+            assert transport.core.context_recovery_candidate(row["request_ref"])
+            # The original provider proof may no longer be readable after
+            # the already-admitted continuation has finished.
+            monkeypatch.setattr(CodexAppServerAdapter, "failed_request_observation", staticmethod(lambda **_: None))
     elif fault == "transport-sync-crash":
         recover = transport.core.recover_failed_request
         def crash_sync(*args, **kwargs):