From af329e23293b0dfad011d10aa2dbc92e616a23e3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 27 Sep 2026 21:32:30 -0500 Subject: [PATCH] fix(chat): close unmatched X reply receipts after recovery window Co-Authored-By: Paperclip --- doc/connections/X.md | 3 ++- .../__tests__/chat-channels.integration.test.ts | 5 +++++ server/src/services/x/service.ts | 17 ++++++++++++++++- 3 files changed, 23 insertions(+), 2 deletions(-) diff --git a/doc/connections/X.md b/doc/connections/X.md index 8e8f66c88ba..6f84af9ed29 100644 --- a/doc/connections/X.md +++ b/doc/connections/X.md @@ -80,7 +80,8 @@ runtime using persisted activation roots. Its automatic `postMessage` path is disabled. Reply targets never depend on the adapter's in-memory latest post. Follow-ups that arrive before their bot-parent link is committed remain in the intake ledger and resume when that exact link becomes available, including -after a restart. Unrelated replies cannot start tasks. +after a restart. Unrelated replies cannot start tasks. Unmatched replies close +after 24 hours with an ignored disposition; their audit receipt is retained. The publication ledger saves exact text and target before any provider request. The same key and payload return the original intent. Reusing a key with a diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 1cad9d986be..62520d4dc23 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -72375,6 +72375,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(f.wakeup).toHaveBeenCalledTimes(3); await f.drain(); expect(f.wakeup).toHaveBeenCalledTimes(3); + await db.update(chatActions).set({ createdAt: new Date(Date.now() - 25 * 60 * 60 * 1000) }).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.providerActionId, "x-post:603"))); + await f.drain(); + const [expired] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.providerActionId, "x-post:603"))); + expect(expired).toMatchObject({ status: "processed", result: { disposition: "ignored", reason: "X reply parent was not recorded within 24 hours" } }); + expect(f.wakeup).toHaveBeenCalledTimes(3); }, 60_000); it("X concurrent reply intents and workers preserve each branch's exact target after restart", async () => { diff --git a/server/src/services/x/service.ts b/server/src/services/x/service.ts index 51e55c67fa0..2547489d65e 100644 --- a/server/src/services/x/service.ts +++ b/server/src/services/x/service.ts @@ -270,6 +270,7 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { : null; } async function processIngress(limit = 25) { + const unmatchedBefore = new Date(Date.now() - 24 * 60 * 60 * 1000); // Unmatched direct replies stay durable without occupying the ready queue. // Once the exact parent link exists, any process can continue the task. const awaitingParentReady = and( @@ -280,6 +281,10 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { sql`${chatActions.payload}->'event'->>'event_type' = 'post.mention.create'`, sql`not exists (select 1 from chat_publications publication where publication.endpoint_id = ${chatActions.endpointId} and publication.state = 'streaming' and publication.payload->'xReply' is not null)`, ), + and( + sql`${chatActions.payload}->'event'->>'event_type' = 'post.reply.create'`, + lte(chatActions.createdAt, unmatchedBefore), + ), ), ); const rows = await db @@ -339,10 +344,20 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { ? event.payload.paperclipThreadId : await resolveThread(endpoint, event); if (!threadId && xParent(event.payload)) { + const expired = + event.event_type === "post.reply.create" && + claimed.createdAt <= unmatchedBefore; await db .update(chatActions) .set({ - status: "awaiting_parent", + status: expired ? "processed" : "awaiting_parent", + result: expired + ? { + claim, + disposition: "ignored", + reason: "X reply parent was not recorded within 24 hours", + } + : { claim }, payload: { ...claimed.payload, parentPostId: xParent(event.payload),