fix(collaboration): preserve original-chat returns across retries

loopx/capabilities/manager_context/presentation.py

@@ -74,7 +74,13 @@ def project_collaboration(
                 or not is_source_chat_host(root, route, store)
             ):
                 continue
-            recovered.setdefault(client, []).append(route)
+            request_id = route.get("request_id")
+            if not isinstance(request_id, str):
+                continue
+            candidates = recovered.setdefault(client, {})
+            # A duplicate route cannot establish this request's lost answer.
+            # Other recipients of the same Turn remain independently readable.
+            candidates[request_id] = route if request_id not in candidates else None
         except (OSError, ValueError, TypeError):
             continue
     result = []
@@ -90,7 +96,10 @@ def project_collaboration(
         if receipts is None:
             receipt = response.get("context_handoff_receipt")
             receipts = (
-                [receipt] if receipt else recovered.get(turn.get("client_turn_id"), [])
+                [receipt] if receipt else [
+                    route for route in recovered.get(turn.get("client_turn_id"), {}).values()
+                    if route is not None
+                ]
             )
         collaborations = []
         for receipt in receipts if isinstance(receipts, list) else []:

loopx/control_plane/collaboration/conversation_binding.ts

@@ -1,6 +1,7 @@
 import type {JsonObject} from "../effect_program.ts";
 import {isDeepStrictEqual} from "node:util";
 import {createHash} from "node:crypto";
+import {BARE_SHA256_PATTERN} from "../content_digest.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {requireInteger, requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts";
 import {requireAutomationCadenceEligibility} from "../quota/automation_cadence.ts";
@@ -611,7 +612,7 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
       && Number.isFinite(created) && created >= start && created <= end
       && Number.isSafeInteger(proof.start_step_index) && Number(proof.start_step_index) >= 0
       && Number.isSafeInteger(proof.total_steps) && Number(proof.total_steps) > Number(proof.start_step_index) + 1
-      && [proof.input_sha256, proof.answer_sha256].every(value => typeof value === "string" && /^[a-f0-9]{64}$/.test(value))
+      && [proof.input_sha256, proof.answer_sha256].every(value => typeof value === "string" && BARE_SHA256_PATTERN.test(value))
       && response?.schema_version === "loopx_chat_agent_response_v0"
       && typeof response.message === "string" && response.message.trim().length > 0
       && (adopted === null || adopted.turn_id === failed.turn_id

loopx/control_plane/collaboration/return_delivery.ts

@@ -142,6 +142,7 @@ interface ReturnCommand {
   key: string;
   path: string;
   now: string;
+  nowMicros: bigint;
   exact: boolean;
   facts: JsonObject;
 }
@@ -152,6 +153,13 @@ function returnAddress(value: unknown): {request_id: string; result_key: string}
     result_key: matchingString(row.result_key, "result_key", RESULT_KEY)};
 }
 
+function returnClockMicros(value: unknown): bigint {
+  const instant = typeof value === "string" && value.length <= 64
+    ? parseTodoTimestampMicros(value) : null;
+  if (instant === null) throw new EffectRuntimeRequestError("invalid return clock");
+  return instant;
+}
+
 function returnCommand(value: unknown): ReturnCommand {
   const request = requireJsonObject(value, "return delivery transaction");
   if (request.schema_version !== RETURN_DELIVERY_TRANSACTION) {
@@ -161,10 +169,10 @@ function returnCommand(value: unknown): ReturnCommand {
   if (!isAbsolute(root)) throw new EffectRuntimeRequestError("return runtime root must be absolute");
   const {request_id: id, result_key: key} = returnAddress(request);
   const now = requireNonEmptyString(request.now, "now");
-  if (now.length > 64 || parseTodoTimestampMicros(now) === null) throw new EffectRuntimeRequestError("invalid return clock");
+  const nowMicros = returnClockMicros(now);
   requireStringLiteral(request.phase, ["prepare", "attempt", "settle"], "return delivery phase");
   const folder = join(root, ".local", "manager-context");
-  return { request, folder, key, now,
+  return { request, folder, key, now, nowMicros,
     path: join(folder, "replies", id, `${key}.delivery.json`),
     exact: requireBoolean(request.exact, "exact"),
     facts: requireJsonObject(request.facts, "return source facts") };
@@ -207,12 +215,12 @@ function unavailableRecord(error: unknown): boolean {
 function pendingDelivery(command: ReturnCommand, state: JsonObject): boolean {
   if (TERMINAL.has(String(state.status))) return false;
   if (!command.exact || state.status !== "admitted") {
-    if (state.retry_at && command.now < String(state.retry_at)) return false;
+    if (state.retry_at && command.nowMicros < returnClockMicros(state.retry_at)) return false;
   }
   const admission = state.admission;
   return !(command.exact && admission && typeof admission === "object"
     && (admission as JsonObject).expires_at
-    && command.now < String((admission as JsonObject).expires_at));
+    && command.nowMicros < returnClockMicros((admission as JsonObject).expires_at));
 }
 
 function deliveryState(value: unknown): JsonObject {
@@ -251,7 +259,7 @@ async function scanReturns(request: JsonObject): Promise<JsonObject> {
 }
 
 function shiftedClock(now: string, seconds: number): string {
-  // Preserve Python's offset/fraction representation, used by existing lexical retry fences.
+  // Preserve Python's offset/fraction representation; eligibility compares instants.
   const match = /^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d)(\.\d+)?(Z|[+-]\d\d:\d\d(?::\d\d(?:\.\d+)?)?)?$/.exec(now);
   if (!match) throw new EffectRuntimeRequestError("invalid return clock representation");
   return new Date(Date.parse(`${match[1]}Z`) + seconds * 1000).toISOString().slice(0, 19)
@@ -379,7 +387,6 @@ function returnAuthorityBlocker(command: ReturnCommand): string | null {
 }
 
 async function prepareReturn(command: ReturnCommand, state: JsonObject): Promise<JsonObject> {
-  if (!pendingDelivery(command, state)) return returnResponse();
   if (command.facts.error) return sourceFailure(command, state);
   let source: Awaited<ReturnType<typeof inspectReturnPublication>>;
   try {
@@ -529,7 +536,10 @@ export async function transactManagerReturnDelivery(value: unknown): Promise<Jso
   const command = returnCommand(value);
   return withFileMutationLock(command.path, async () => {
     let state: JsonObject;
-    try { state = deliveryState(await readReturnRecord(command.path, {})); }
+    try {
+      state = deliveryState(await readReturnRecord(command.path, {}));
+      if (command.request.phase === "prepare" && !pendingDelivery(command, state)) return returnResponse();
+    }
     catch (error) {
       if (command.request.phase === "prepare" && unavailableRecord(error)) return returnResponse();
       throw error;

tests/conftest.py

@@ -91,6 +91,8 @@ def _isolated_default_runtime_routes(tmp_path_factory, monkeypatch):
     monkeypatch.setenv("HOME", str(home))
     monkeypatch.setenv("USERPROFILE", str(home))
     monkeypatch.delenv("CODEX_HOME", raising=False)
+    # Tests opt into provider selection; isolate the operator's Chat route.
+    monkeypatch.delenv("LOOPX_CHAT_CODEX_MODEL_PROVIDER", raising=False)
     for module_name, attribute, parts in _DEFAULT_ROUTE_REFERENCES:
         monkeypatch.setattr(importlib.import_module(module_name), attribute, home.joinpath(*parts))
     yield

tests/control_plane_ts/content_digest_single_owner.test.ts

@@ -108,6 +108,7 @@ const CANONICAL_CONSUMERS = [
   "control_plane/capabilities/trajectory_input_usage.ts",
   "control_plane/capabilities/trajectory_tool_context.ts",
   "control_plane/collaboration/chat_mode.ts",
+  "control_plane/collaboration/conversation_binding.ts",
   "control_plane/collaboration/delegation.ts",
   "control_plane/collaboration/delegation_result_use.ts",
   "control_plane/collaboration/goal_instance_lifecycle.ts",

tests/control_plane_ts/conversation_binding.test.ts

@@ -45,6 +45,16 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
   assert.equal(plan.operation, "adopt_native_result");
   assert.equal(plan.recovery_eligible, true);
   assert.equal(plan.client_turn_id, undefined);
+  for (const field of ["input_sha256", "answer_sha256"]) {
+    for (const invalid of [null, false, 1, [], {}, "", "1".repeat(63), "1".repeat(65),
+      "A".repeat(64), `sha256:${"1".repeat(64)}`]) {
+      assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery,
+        native_observation: {...proof, [field]: invalid}}}).recovery_eligible, false,
+      `${field} must reject ${JSON.stringify(invalid)}`);
+    }
+    assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery,
+      native_observation: {...proof, [field]: "abcdef0123456789".repeat(4)}}}).recovery_eligible, true);
+  }
   for (const change of [{source_verified: false}, {stop_observed: true}, {steering_observed: true},
     {attachments_present: true}, {native_observation: {...proof, thread_id: "other"}},
     {native_observation: {...proof, request_ref: "a".repeat(24)}},

tests/control_plane_ts/manager_return_transaction.test.ts

@@ -146,6 +146,90 @@ test("an ambiguous expired external admission never dispatches another send", as
   } finally { await f.close(); }
 });
 
+const deadlineCases = [
+  {label: "offset before", now: "2026-01-02T04:00:00+01:00", deadline: "2026-01-02T03:30:00+00:00", due: false},
+  {label: "offset due", now: "2026-01-02T03:30:00Z", deadline: "2026-01-02T04:00:00+01:00", due: true},
+  {label: "microsecond before", now: "2026-01-02T03:30:00Z", deadline: "2026-01-02T03:30:00.000001+00:00", due: false},
+  {label: "fraction precision before", now: "2026-01-02T03:30:00.1Z", deadline: "2026-01-02T03:30:00.100001+00:00", due: false},
+  {label: "equal instant", now: "2026-01-02T03:30:00+01:00", deadline: "2026-01-02T02:30:00Z", due: true},
+  {label: "offset seconds due", now: "2026-01-02T03:30:00.123457Z", deadline: "2026-01-02T04:30:30.123456+01:00:30", due: true},
+];
+
+for (const clock of deadlineCases) {
+  for (const exact of [false, true]) {
+    test(`return retry clock uses instants: ${clock.label} (${exact})`, async () => {
+      const f = await fixture(exact, true);
+      try {
+        await f.put(f.statePath, {status: "retry_pending", retry_at: clock.deadline});
+        const bytes = await readFile(f.statePath);
+        const scanned = await f.run("scan", {now: clock.now,
+          candidates: [{request_id: id, result_key: "conclusion"}]});
+        assert.deepEqual(scanned.candidates, clock.due ? [{request_id: id, result_key: "conclusion"}] : []);
+        assert.equal(scanned.unavailable_count, 0);
+        assert.deepEqual(await readFile(f.statePath), bytes);
+        assert.equal((await f.run("prepare", {now: clock.now})).action, clock.due ? "send" : "skip");
+        if (!clock.due) assert.deepEqual(await readFile(f.statePath), bytes);
+      } finally { await f.close(); }
+    });
+  }
+
+  test(`return admission clock retains the original attempt: ${clock.label}`, async () => {
+    const f = await fixture(true, true);
+    try {
+      await f.put(f.statePath, {status: "admitted", goal_ref: ref, attempt,
+        admission: {token: "prior-send", prior_status: "verification_required", expires_at: clock.deadline}});
+      const bytes = await readFile(f.statePath);
+      const scanned = await f.run("scan", {now: clock.now,
+        candidates: [{request_id: id, result_key: "conclusion"}]});
+      assert.deepEqual(scanned.candidates, clock.due ? [{request_id: id, result_key: "conclusion"}] : []);
+      assert.deepEqual(await readFile(f.statePath), bytes);
+      const plan = await f.run("prepare", {now: clock.now});
+      assert.equal(plan.action, clock.due ? "verify" : "skip");
+      if (clock.due) {
+        assert.deepEqual(plan.attempt, attempt);
+        await f.run("settle", {...plan, now: clock.now, outcome: {kind: "verified",
+          verification: {verification_performed: true, reply_verified: true}}});
+        assert.equal((await f.state()).status, "delivered");
+        assert.equal((await f.state()).provider_receipt, attempt.provider_receipt);
+      } else {
+        assert.deepEqual(await readFile(f.statePath), bytes);
+      }
+    } finally { await f.close(); }
+  });
+}
+
+for (const field of ["retry_at", "expires_at"]) {
+  for (const malformed of ["not-a-clock", "2026-02-30T03:04:05Z", 42, {}]) {
+    test(`return clock damage is unavailable without rewriting ${field}: ${JSON.stringify(malformed)}`, async () => {
+      const f = await fixture(true, true);
+      try {
+        await f.put(f.statePath, field === "retry_at"
+          ? {status: "retry_pending", retry_at: malformed}
+          : {status: "admitted", admission: {expires_at: malformed}, attempt});
+        const bytes = await readFile(f.statePath);
+        const scanned = await f.run("scan", {candidates: [{request_id: id, result_key: "conclusion"}]});
+        assert.deepEqual(scanned.candidates, []);
+        assert.equal(scanned.unavailable_count, 1);
+        assert.equal((await f.run("prepare")).action, "skip");
+        assert.deepEqual(await readFile(f.statePath), bytes);
+      } finally { await f.close(); }
+    });
+  }
+}
+
+test("return clock admission rechecks a due scan after another writer defers it", async () => {
+  const f = await fixture(true, true);
+  try {
+    const candidates = [{request_id: id, result_key: "conclusion"}];
+    await f.put(f.statePath, {status: "retry_pending", retry_at: "2026-01-02T03:00:00Z"});
+    assert.deepEqual((await f.run("scan", {candidates})).candidates, candidates);
+    await f.put(f.statePath, {status: "retry_pending", retry_at: "2026-01-02T04:04:05.123457+01:00"});
+    const bytes = await readFile(f.statePath);
+    assert.equal((await f.run("prepare")).action, "skip");
+    assert.deepEqual(await readFile(f.statePath), bytes);
+  } finally { await f.close(); }
+});
+
 test("invalid transaction addresses and phases refuse before changing evidence", async () => {
   const f = await fixture(false);
   try {