diff --git a/.github/workflows/storybook-visual.yml b/.github/workflows/storybook-visual.yml index 4b3773615c1..97fa5bfa1ac 100644 --- a/.github/workflows/storybook-visual.yml +++ b/.github/workflows/storybook-visual.yml @@ -46,6 +46,8 @@ jobs: steps: - name: Checkout repository uses: actions/checkout@v7 + with: + persist-credentials: false - name: Setup pnpm uses: pnpm/action-setup@v6 @@ -59,7 +61,13 @@ jobs: cache: pnpm - name: Install dependencies - run: pnpm install --frozen-lockfile + # CI owns committed lockfile updates. Resolve dependency-changing PRs + # locally, without running dependency lifecycle scripts. + run: | + if ! pnpm install --ignore-scripts --frozen-lockfile; then + pnpm install --resolution-only --ignore-scripts --no-frozen-lockfile + pnpm install --ignore-scripts --frozen-lockfile + fi - name: Install Playwright Chromium run: npx playwright install --with-deps chromium diff --git a/doc/connections/X.md b/doc/connections/X.md index bc42d289c15..8e8f66c88ba 100644 --- a/doc/connections/X.md +++ b/doc/connections/X.md @@ -78,10 +78,14 @@ Signed events enter a durable intake ledger before acknowledgement. The patched `@chat-adapter/x@4.39.0` normalizes mention and reply events into the shared chat 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. The publication ledger saves exact text and target before any provider request. -The same key returns the original intent, and the same interaction cannot gain a -second intent. A worker crash or ambiguous network error becomes +The same key and payload return the original intent. Reusing a key with a +different target or text returns a conflict, and the same interaction cannot +gain a second intent. A worker crash or ambiguous network error becomes `delivery_unknown`; it is never automatically replayed. Inspect X and Activity, then mark delivered or cancel. Marking delivered does not invent a missing post ID or restore follow-up routing to an unknown post. Definite rejections, expired diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 9803b78fc3e..378f91cac6f 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -72246,6 +72246,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const botId = String(Date.now()); const sends: Array> = []; let sendFailure = false; + let beforeSendResponse: (() => Promise) | null = null; const providerFetch = vi.fn(async (url: string | URL | Request, init?: RequestInit) => { if (String(url).endsWith("/2/oauth2/token")) { const params = new URLSearchParams(String(init?.body)); @@ -72256,6 +72257,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { if (String(url).endsWith("/2/tweets")) { sends.push(JSON.parse(String(init?.body))); if (sendFailure) throw new Error("connection closed after acceptance"); + await beforeSendResponse?.(); return Response.json({ data: { id: String(900000 + sends.length) } }); } return new Response(null, { status: 404 }); @@ -72274,10 +72276,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const body = JSON.stringify({ data }); return service.handleWebhook(endpoint.publicId, "x", new Request(`https://paperclip.example/api/chat-webhooks/${endpoint.publicId}/x`, { method: "POST", body, headers: { "content-type": "application/json", "x-twitter-webhooks-signature-oauth2": xSignature(body, valid ? "x-client-secret" : "wrong") } })); } - async function drain() { for (let i = 0; i < 3; i++) await service.processPendingDeliveries(); + async function drain() { + for (let i = 0; i < 3; i++) await service.processPendingDeliveries(); } async function binding(id: string, responsibleUserId: string | null = "owner-user") { const [link] = await db.select().from(chatMessageLinks).where(and(eq(chatMessageLinks.endpointId, endpoint.id), eq(chatMessageLinks.providerMessageId, id))); + expect(link, `X post ${id} must have a durable task binding`).toBeDefined(); const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.id, link.conversationId)); const [action] = await db.select().from(chatActions).where(and(eq(chatActions.deliveryId, link.deliveryId!), eq(chatActions.kind, "inbound_wakeup"))); const runId = randomUUID(); @@ -72287,6 +72291,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } return { ...fixture, endpoint, botId, sends, event, deliver, drain, binding, wakeup, principal, providerFetch, service: () => service, failSend: () => { sendFailure = true; }, + beforeSendResponse: (hook: () => Promise) => { beforeSendResponse = hook; }, restart: async () => { await service.shutdown(); service = chatChannelService(db, options); fixtureServices.add(service); }, }; } @@ -72323,6 +72328,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const result = await native.execute({ tool: "x_reply", callId: randomUUID(), arguments: request }) as { publicationId: string; status: string }; expect(result.status).toBe("pending"); expect(await executeConnectorTool(db, binding, "x_reply", request)).toEqual(result); + await expect(executeConnectorTool(db, binding, "x_reply", { ...request, text: "Different intent" })).rejects.toThrow("different X reply"); await expect(executeConnectorTool(db, binding, "x_reply", { ...request, idempotencyKey: randomUUID() })).rejects.toThrow("already has a reply"); await issueService(db).addComment(binding.issueId, "Internal final response", { agentId: binding.agentId }, { authorType: "agent" }); await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, binding.runId)); @@ -72337,9 +72343,39 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const links = await db.select().from(chatMessageLinks).where(eq(chatMessageLinks.endpointId, f.endpoint.id)); expect(links.find(l => l.providerMessageId === "202")?.conversationId).toBe(links.find(l => l.providerMessageId === "200")?.conversationId); expect(links.find(l => l.providerMessageId === "203")?.conversationId).not.toBe(links.find(l => l.providerMessageId === "200")?.conversationId); + const followup = await f.binding("202"); + await expect(executeConnectorTool(db, followup, "x_reply", { ...request, replyToPostId: "202" })).rejects.toThrow("different X reply"); expect(f.sends).toHaveLength(1); }, 60_000); + it("X retains early follow-ups until their outbound parent link commits", async () => { + const f = await configuredXEndpoint(); + const { executeConnectorTool } = await import("../services/connector-runtime.js"); + await f.deliver(f.event("600")); await f.drain(); + const binding = await f.binding("600"); + await executeConnectorTool(db, binding, "x_reply", { replyToPostId: "600", text: "Reply accepted by X", idempotencyKey: randomUUID() }); + f.beforeSendResponse(async () => { + // X knows the bot's post ID, but the send response has not reached us. + await f.deliver(f.event("601", { in_reply_to_tweet_id: "900001" }, "post.reply.create")); + await f.deliver(f.event("602", { in_reply_to_tweet_id: "900001" })); + await f.deliver(f.event("603", { in_reply_to_tweet_id: "123456" }, "post.reply.create")); + await f.drain(); + expect(f.wakeup).toHaveBeenCalledOnce(); + const waiting = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "x_webhook_ingress"), eq(chatActions.status, "awaiting_parent"))); + expect(waiting).toHaveLength(3); + }); + await f.service().processPendingPublications(); + await f.restart(); await f.drain(); + const links = await db.select().from(chatMessageLinks).where(eq(chatMessageLinks.endpointId, f.endpoint.id)); + const root = links.find(link => link.providerMessageId === "600")!; + expect(links.find(link => link.providerMessageId === "601")?.conversationId).toBe(root.conversationId); + expect(links.find(link => link.providerMessageId === "602")?.conversationId).toBe(root.conversationId); + expect(links.some(link => link.providerMessageId === "603")).toBe(false); + expect(f.wakeup).toHaveBeenCalledTimes(3); + await f.drain(); + expect(f.wakeup).toHaveBeenCalledTimes(3); + }, 60_000); + it("X concurrent reply intents and workers preserve each branch's exact target after restart", async () => { const f = await configuredXEndpoint(); const { executeConnectorTool } = await import("../services/connector-runtime.js"); @@ -72364,6 +72400,26 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ])); }, 60_000); + it("X restart recovery passes retained receipts from paused and attention endpoints", async () => { + const f = await configuredXEndpoint(); + for (const status of ["paused", "attention"] as const) { + const other = await f.service().create(f.companyId, { provider: "x", assignedAgentId: f.assignedAgentId }, "owner-user"); + await db.update(chatEndpoints).set({ status }).where(eq(chatEndpoints.id, other.id)); + await db.insert(chatDeliveries).values(Array.from({ length: 25 }, (_, index) => ({ + companyId: f.companyId, endpointId: other.id, + providerEventId: `${status}-${index}`, deduplicationKey: `${status}-${index}`, + eventKind: "message" as const, normalizedEvent: { conversation: { externalThreadId: `retained-${index}` } }, + state: "received" as const, receivedAt: new Date("2026-01-01T00:00:00Z"), + }))); + } + await f.deliver(f.event("700")); + await f.restart(); await f.drain(); + expect(f.wakeup).toHaveBeenCalledOnce(); + await f.binding("700"); + const retained = await db.select().from(chatDeliveries).where(and(eq(chatDeliveries.companyId, f.companyId), eq(chatDeliveries.state, "received"))); + expect(retained).toHaveLength(50); + }, 60_000); + it("X opt-out and current authority stop work and sends; uncertain sends never retry", async () => { const f = await configuredXEndpoint(); const { executeConnectorTool } = await import("../services/connector-runtime.js"); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index b2033bfab31..e19c75ab891 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -27849,7 +27849,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { .from(chatDeliveries) .where( and( - sql`not exists (select 1 from chat_endpoints e where e.id = ${chatDeliveries.endpointId} and e.provider = 'agentmail')`, + // These endpoints cannot drain. Keep their receipts durable, but + // do not let an old paused backlog starve active connections. + sql`not exists (select 1 from chat_endpoints e where e.id = ${chatDeliveries.endpointId} and (e.provider = 'agentmail' or e.status in ('paused', 'attention')))`, onlyDeliveryId ? eq(chatDeliveries.id, onlyDeliveryId) : undefined, notInArray(chatDeliveries.eventKind, [ "reaction_added", diff --git a/server/src/services/x/service.ts b/server/src/services/x/service.ts index f8ac2818950..51e55c67fa0 100644 --- a/server/src/services/x/service.ts +++ b/server/src/services/x/service.ts @@ -248,6 +248,20 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { ) .limit(1); if (parent) return parent.thread; + // X can deliver a follow-up before the send response and outbound link + // commit. Do not activate a second branch during that publication window. + const [sending] = await db + .select({ id: chatPublications.id }) + .from(chatPublications) + .where( + and( + eq(chatPublications.endpointId, endpoint.id), + eq(chatPublications.state, "streaming"), + sql`${chatPublications.payload}->'xReply' is not null`, + ), + ) + .limit(1); + if (sending) return null; } // An explicit new invocation branches at this post, regardless of the // original X conversation. Unrelated replies are never wakeups. @@ -256,6 +270,18 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { : null; } async function processIngress(limit = 25) { + // 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( + eq(chatActions.status, "awaiting_parent"), + or( + sql`exists (select 1 from chat_message_links link where link.endpoint_id = ${chatActions.endpointId} and link.direction = 'outbound' and link.provider_message_id = ${chatActions.payload}->>'parentPostId')`, + and( + 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)`, + ), + ), + ); const rows = await db .select() .from(chatActions) @@ -264,6 +290,7 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { eq(chatActions.kind, "x_webhook_ingress"), or( eq(chatActions.status, "received"), + awaitingParentReady, and( eq(chatActions.status, "processing"), lte(chatActions.updatedAt, new Date(Date.now() - 120_000)), @@ -283,6 +310,7 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { eq(chatActions.id, action.id), or( eq(chatActions.status, "received"), + awaitingParentReady, and( eq(chatActions.status, "processing"), lte(chatActions.updatedAt, new Date(Date.now() - 120_000)), @@ -310,6 +338,25 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { typeof event.payload.paperclipThreadId === "string" ? event.payload.paperclipThreadId : await resolveThread(endpoint, event); + if (!threadId && xParent(event.payload)) { + await db + .update(chatActions) + .set({ + status: "awaiting_parent", + payload: { + ...claimed.payload, + parentPostId: xParent(event.payload), + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + sql`${chatActions.result}->>'claim' = ${claim}`, + ), + ); + continue; + } if (threadId) { const context = event.payload.paperclipContext ?? @@ -444,7 +491,18 @@ export function xChannelService(db: Db, hooks: XHooks, fetchImpl = fetch) { eq(chatPublications.idempotencyKey, key), ), ); - if (existing) return existing; + if (existing) { + if ( + existing.payload.xReply?.replyToPostId !== request.replyToPostId || + existing.payload.text !== request.text + ) { + throw conflict( + "This idempotency key belongs to a different X reply; use a new key for a new interaction", + { publicationId: existing.id }, + ); + } + return existing; + } const [interaction] = await tx .select() .from(chatPublications)