fix(weixin): retain retired account receipts and enforce chat thread ownership

Require a durable account-and-chat binding before /resume reads a thread, and
filter thread summaries to that chat's known bindings. Keep bounded binding
history so /new does not strand a chat's own earlier conversations. A replacement
account's explicit /new durably retains the previous account state for review
and starts fresh without resubmitting its pending prompt or uncertain reply.
Same-account pending work still blocks replacement and now receives an answer.

Failed batch handlers retain their inflight claims after rejection, matching the
production poller. Existing pending admission preservation is retained and
proved through a real restarted process at the durable thread-binding boundary.

Focused Node bridge gate: 48 passed, 0 failed, 0 skipped, including five new
process-level ownership, retirement and restart cases. Evidence:
artifacts/takeover-0101-20261005/Weixin-authority-recovery-final-gate.log
Hosted CI, final combined Engine gates and installed release QA remain separate.

integrations/weixin-bridge/src/index.mjs

@@ -274,8 +274,12 @@ async function handleCommand(chatId, command, inbound = {}) {
       await sendThreads(chatId);
       return;
     case "new_thread": {
-      const state = await ensureThread(chatId, { forceNew: true });
-      await sendText(chatId, `Created thread ${state.threadId}`);
+      try {
+        const state = await ensureThread(chatId, { forceNew: true });
+        await sendText(chatId, `Created thread ${state.threadId}`);
+      } catch (error) {
+        await sendText(chatId, `New conversation failed: ${error.message}`);
+      }
       return;
     }
     case "resume":
@@ -302,13 +306,24 @@ async function handleCommand(chatId, command, inbound = {}) {
 }
 
 async function ensureThread(chatId, { forceNew = false, threadRequest } = {}) {
-  const existing = await threadStore.getChat(chatId);
-  if (forceNew && hasPendingWork(existing)) throw new Error("This chat still has a pending turn or retained reply. Use /status first.");
+  let existing = await threadStore.getChat(chatId);
+  const previousAccounts = new Set([existing?.bindingAccountId, existing?.pendingAdmission?.accountId, existing?.turnDelivery?.accountId].filter(Boolean));
+  const replacingAccount = forceNew && previousAccounts.size === 1 && !previousAccounts.has(accountIdentity());
+  if (forceNew && hasPendingWork(existing) && !replacingAccount) throw new Error("This chat still has a pending turn or retained reply. Use /status first.");
   if (existing?.threadId && !forceNew) {
     if (existing.bindingAccountId !== accountIdentity()) throw new Error("This thread belongs to another or unverified bot account. Use /new to create an explicit binding.");
     return existing;
   }
 
+  if (replacingAccount) {
+    // Explicit /new retires the previous account's private receipt before any
+    // new side effect. It remains in this store, outside automatic recovery.
+    const { retiredAccountStates = {}, ...retired } = existing;
+    const receipt = crypto.createHash("sha256").update(JSON.stringify(retired)).digest("hex");
+    await threadStore.patchChat(chatId, { retiredAccountStates: { ...retiredAccountStates, [receipt]: retired } });
+    existing = await threadStore.getChat(chatId);
+  }
+
   const effectiveModel = existing?.model || config.model;
 
   const thread = await runtimeJson("/v1/threads", {
@@ -327,7 +342,10 @@ async function ensureThread(chatId, { forceNew = false, threadRequest } = {}) {
   });
 
   const state = {
-    ...preservedChatStateFields(existing, ["model", "authorizedIdentity", "contextToken", "pendingAdmission"]),
+    ...preservedChatStateFields(existing, replacingAccount
+      ? ["model", "contextToken", "retiredAccountStates"]
+      : ["model", "authorizedIdentity", "contextToken", "pendingAdmission", "retiredAccountStates"]),
+    threadBindings: [...threadBindings(chatId, existing), { threadId: thread.id, accountId: accountIdentity(), actorId: turnActor(chatId) }].slice(-threadStore.options.actionLimit),
     threadId: thread.id,
     bindingAccountId: accountIdentity(),
     lastSeq: 0,
@@ -346,6 +364,19 @@ function turnActor(chatId) {
   return accountIdentity() && chatId ? `weixin-account:${accountIdentity()}:user:${chatId}` : "";
 }
 
+function threadBindings(chatId, state) {
+  const bindings = (Array.isArray(state?.threadBindings) ? state.threadBindings : []).filter((entry) => entry?.threadId && entry.accountId && entry.actorId);
+  if (state?.threadId && state.bindingAccountId) {
+    const current = { threadId: state.threadId, accountId: state.bindingAccountId, actorId: `weixin-account:${state.bindingAccountId}:user:${chatId}` };
+    if (!bindings.some((entry) => entry.threadId === current.threadId && entry.accountId === current.accountId && entry.actorId === current.actorId)) bindings.push(current);
+  }
+  return bindings.slice(-threadStore.options.actionLimit);
+}
+
+function ownsThread(chatId, state, threadId) {
+  return threadBindings(chatId, state).some((entry) => entry.threadId === threadId && entry.accountId === accountIdentity() && entry.actorId === turnActor(chatId));
+}
+
 function inboundKey(msg) {
   const id = msg.message_id;
   if (!accountIdentity() || !msg.from_user_id || id === undefined || id === null || String(id).trim() === "") return null;
@@ -609,7 +640,7 @@ async function sendStatus(chatId) {
       [
         `user_id=${chatId}`,
         state?.pendingAdmission ? "Turn admission is pending; its original request is retained." : "",
-        state?.activeTurnId ? `accepted_turn=${state.activeTurnId}` : "",
+        state?.activeTurnId && ownsRecovery(chatId, state) ? `accepted_turn=${state.activeTurnId}` : "",
         state?.turnDelivery?.outputs?.some((entry) => entry.status === "uncertain") ? "A reply has unconfirmed API acceptance. It is retained for review and will not be resent automatically." : "",
         `runtime=${health.status || "unknown"}`,
         `version=${runtimeInfo.version || "unknown"}`,
@@ -630,11 +661,12 @@ async function sendStatus(chatId) {
 
 async function sendThreads(chatId) {
   try {
-    const threads = await runtimeJson(
+    const state = await threadStore.getChat(chatId);
+    const threads = (await runtimeJson(
       "/v1/threads/summary?limit=8&include_archived=true"
-    );
+    )).filter((thread) => ownsThread(chatId, state, thread.id));
     if (!threads.length) {
-      await sendText(chatId, "No runtime threads yet.");
+      await sendText(chatId, "No recent runtime threads owned by this account and chat.");
       return;
     }
     await sendText(
@@ -660,11 +692,13 @@ async function resumeThread(chatId, args) {
   try {
     const existing = await threadStore.getChat(chatId);
     if (hasPendingWork(existing)) throw new Error("This chat still has a pending turn or retained reply. Use /status first.");
+    if (!ownsThread(chatId, existing, threadId)) throw new ApprovalOwnershipError("Thread is not owned by this account and chat");
     const detail = await runtimeJson(
       `/v1/threads/${encodeURIComponent(threadId)}`
     );
     await threadStore.setChat(chatId, {
-      ...preservedChatStateFields(existing, ["model", "authorizedIdentity", "contextToken"]),
+      ...preservedChatStateFields(existing, ["model", "authorizedIdentity", "contextToken", "retiredAccountStates"]),
+      threadBindings: threadBindings(chatId, existing),
       threadId,
       bindingAccountId: accountIdentity(),
       lastSeq: Number(detail.latest_seq || 0),

integrations/weixin-bridge/src/lib.mjs

@@ -47,17 +47,15 @@ export async function processUpdateBatch({ messages, nextCursor, store, keyOf, h
   for (const msg of messages || []) {
     const key = keyOf(msg);
     if (!key) continue;
-    const claim = await store.claimMessage(key);
+    const claim = store.data.inflight?.[key] ? "interrupted" : await store.claimMessage(key);
     if (claim === "done") continue;
     if (claim === "interrupted") {
       await interrupted(msg);
+      await store.completeMessage(key);
       continue;
     }
-    try {
-      await handle(msg);
-    } finally {
-      await store.completeMessage(key);
-    }
+    await handle(msg);
+    await store.completeMessage(key);
   }
   if (nextCursor) await commitCursor(nextCursor);
 }

integrations/weixin-bridge/test/lib.test.mjs

@@ -77,8 +77,6 @@ test("a crash mid-batch replays the batch without losing or re-running prompts",
       messages: batch, nextCursor: "buf-2", store: first, keyOf,
       handle: async (msg) => {
         if (msg.message_id === 2) {
-          // What a restarted process finds on disk while message 2 runs.
-          afterCrash = await ThreadStore.open(statePath, { messageLimit: 50 });
           throw crash;
         }
         handled.push(msg.message_id);
@@ -86,6 +84,8 @@ test("a crash mid-batch replays the batch without losing or re-running prompts",
       interrupted: async () => assert.fail("nothing is interrupted on the first pass"),
       commitCursor: async (buf) => cursors.push(buf),
     }), crash);
+    afterCrash = await ThreadStore.open(statePath, { messageLimit: 50 });
+    assert.ok(afterCrash.data.inflight["u:2"], "a failed handler must remain inflight on disk after rejection");
     assert.deepEqual(cursors, [], "the cursor does not move past unhandled messages");
 
     const reported = [];

integrations/weixin-bridge/test/runtime.test.mjs

@@ -19,7 +19,7 @@ function incoming(id, text, user = "alice") {
 }
 async function fixture(t, options = {}) {
   const dir = await fs.mkdtemp(path.join(os.tmpdir(), "weixin-runtime-"));
-  const state = { polls: 0, batches: [], posts: [], sent: [], lookups: 0, streams: new Set(), turns: [], items: [], seq: 0, approvals: [], decisions: [], logs: "" };
+  const state = { polls: 0, batches: [], posts: [], sent: [], lookups: 0, threadCreates: 0, threadReads: [], streams: new Set(), turns: [], items: [], seq: 0, approvals: [], decisions: [], logs: "" };
   const json = (response, body, status = 200) => { response.writeHead(status, { "Content-Type": "application/json" }); response.end(JSON.stringify(body)); };
   const server = http.createServer(async (req, res) => {
     let raw = ""; for await (const chunk of req) raw += chunk;
@@ -42,9 +42,13 @@ async function fixture(t, options = {}) {
     if (url.pathname === "/v1/workspace/status") return json(res, { workspace: dir });
     if (url.pathname === "/v1/threads" && req.method === "POST") {
       state.threadRequest = body;
-      return json(res, { id: "thread-1" });
+      return json(res, { id: `thread-${++state.threadCreates}` });
+    }
+    if (url.pathname === "/v1/threads/summary") return json(res, options.summaries || []);
+    if (/^\/v1\/threads\/[^/]+$/.test(url.pathname)) {
+      state.threadReads.push(url.pathname);
+      return json(res, { id: url.pathname.split("/").at(-1), latest_seq: state.seq, turns: state.turns, items: state.items, pending_approvals: state.approvals });
     }
-    if (url.pathname === "/v1/threads/thread-1") return json(res, { id: "thread-1", latest_seq: state.seq, turns: state.turns, items: state.items, pending_approvals: state.approvals });
     if (url.pathname === "/v1/threads/thread-1/turns") {
       state.posts.push(body);
       const turn = { id: "turn-1", thread_id: "thread-1", status: "in_progress" };
@@ -211,3 +215,72 @@ test("unanswered admission response leaves polling and status controls live", as
   assert.ok(Date.now() - started < 7000); assert.equal(f.posts.length, 1);
   assert.equal((await f.disk()).chats.alice.pendingAdmission.submitted, true);
 });
+
+test("replacement account explicitly starts fresh while retaining the old uncertain receipt", async (t) => {
+  const f = await fixture(t);
+  const old = { threadId: "old-thread", bindingAccountId: "bot-A", activeTurnId: "old-turn", contextToken: "old-context", turnDelivery: {
+    accountId: "bot-A", actorId: "weixin-account:bot-A:user:alice", messageKey: "old-key", turnId: "old-turn",
+    outputs: [{ text: "old private answer", status: "uncertain", clientId: "old-client" }],
+  } };
+  await fs.writeFile(path.join(f.dir, "thread-map.json"), JSON.stringify({ chats: { alice: old }, messages: [], inflight: {} }));
+  await fs.writeFile(path.join(f.dir, "account.json"), JSON.stringify({ ...f.account, accountId: "bot-B" }));
+  f.batches.push([incoming(1, "/new")]); f.start();
+  await until(() => f.sent.some((msg) => msg.item_list[0].text_item.text === "Created thread thread-1"), "replacement account remained blocked");
+  const current = (await f.disk()).chats.alice;
+  assert.equal(current.bindingAccountId, "bot-B"); assert.equal(current.activeTurnId, null); assert.equal(current.turnDelivery, undefined);
+  const retained = Object.values(current.retiredAccountStates);
+  assert.equal(retained.length, 1); assert.equal(retained[0].threadId, "old-thread");
+  assert.deepEqual(retained[0].turnDelivery, old.turnDelivery); assert.equal(retained[0].activeTurnId, "old-turn");
+  await delay(1100); assert.equal(f.posts.length, 0); assert.equal(f.threadReads.length, 0);
+  assert.equal(f.sent.some((msg) => msg.item_list[0].text_item.text.includes("old private answer")), false);
+});
+
+test("same-account new-thread command cannot discard a pending admission", async (t) => {
+  const f = await fixture(t, { admission: () => {} }); f.batches.push([incoming(1, "pending")]); f.start();
+  await until(() => f.posts.length === 1, "pending admission missing");
+  f.batches.push([incoming(2, "/new")]);
+  await until(() => f.sent.some((msg) => msg.item_list[0].text_item.text.includes("New conversation failed: This chat still has a pending turn or retained reply")), "same-account pending state was overwritten or the control gave no answer");
+  assert.equal(f.threadCreates, 1); assert.equal((await f.disk()).chats.alice.pendingAdmission.request.prompt, "pending");
+});
+
+test("resume and thread summaries retain account-and-chat ownership across new conversations", async (t) => {
+  const f = await fixture(t, { summaries: [
+    { id: "thread-1", title: "Alice private title" }, { id: "unbound-thread", title: "Other account private title" },
+  ] });
+  f.batches.push([incoming(1, "/new"), incoming(2, "/resume thread-1", "bob"), incoming(3, "/threads", "bob"), incoming(4, "/threads")]);
+  f.start();
+  await until(() => f.sent.length === 4, "ownership controls were not handled");
+  const texts = f.sent.map((msg) => msg.item_list[0].text_item.text);
+  assert.match(texts[1], /not owned by this account and chat/);
+  assert.match(texts[2], /No recent runtime threads owned/);
+  assert.match(texts[3], /Alice private title/); assert.ok(!texts.some((text) => text.includes("Other account private title")));
+  assert.equal(f.threadReads.length, 0, "rejected resume must not read another chat's thread");
+  assert.equal((await f.disk()).chats.bob.threadId, undefined);
+  f.batches.push([incoming(5, "/new"), incoming(6, "/resume thread-1")]);
+  await until(() => f.sent.some((msg) => msg.item_list[0].text_item.text === "Resumed thread thread-1"), "owner could not resume its prior conversation");
+  assert.equal((await f.disk()).chats.alice.threadId, "thread-1"); assert.equal(f.threadCreates, 2); assert.equal(f.threadReads.length, 1);
+});
+
+test("replacement account cannot resume the retired account's owned thread", async (t) => {
+  const f = await fixture(t);
+  await fs.writeFile(path.join(f.dir, "thread-map.json"), JSON.stringify({ chats: { alice: { threadId: "thread-1", bindingAccountId: "bot-A" } }, messages: [], inflight: {} }));
+  await fs.writeFile(path.join(f.dir, "account.json"), JSON.stringify({ ...f.account, accountId: "bot-B" }));
+  f.batches.push([incoming(1, "/resume thread-1")]); f.start();
+  await until(() => f.sent.some((msg) => msg.item_list[0].text_item.text.includes("not owned by this account and chat")), "foreign account resumed a private conversation");
+  assert.equal(f.threadReads.length, 0); assert.equal((await f.disk()).chats.alice.bindingAccountId, "bot-A");
+});
+
+test("restart between durable thread binding and admission update keeps the frozen prompt", async (t) => {
+  const f = await fixture(t); const key = JSON.stringify(["bot-A", "alice", "1"]);
+  const request = { prompt: "durable prompt", input_summary: "durable prompt", model: "frozen-model", mode: "agent", allow_shell: false, trust_mode: false, auto_approve: false, operation_key: "frozen-operation" };
+  await fs.writeFile(path.join(f.dir, "thread-map.json"), JSON.stringify({ chats: { alice: {
+    threadId: "thread-1", bindingAccountId: "bot-A", pendingAdmission: {
+      accountId: "bot-A", actorId: "weixin-account:bot-A:user:alice", messageKey: key, request,
+      operationKey: request.operation_key, threadId: null, submitted: false, sinceSeq: null,
+    },
+  } }, messages: [key], inflight: { [key]: "interrupted" } }));
+  f.batches.push([incoming(1, "durable prompt")]); f.start();
+  await until(() => f.streams.size, "durable thread binding lost its pending prompt on restart");
+  assert.equal(f.threadCreates, 0); assert.equal(f.posts.length, 1); assert.deepEqual(f.posts[0], request);
+  assert.equal((await f.disk()).chats.alice.turnDelivery.messageKey, key);
+});