Files
PaperClipAI/server/src/__tests__/chat-interaction-publications.test.ts
T
DottaandPaperclip 19e9facc5b perf(chat): wake delivery queues from commits and deadlines (#15802)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Chat connectors store incoming messages and outgoing publications in
durable queues.
> - The app checks these queues every second, even when they are empty.
> - Empty scans keep the database active and hinder instance sleep.
> - This pull request wakes three queue workers after transactions
settle and at known retry or recovery deadlines.
> - Empty queues stop scheduling checks. Existing rows retain their
ordering, retry, and recovery rules.

## Linked Issues or Issue Description

**What existing behavior does this improve?**

Idle database activity from incoming chat deliveries, outgoing
publications, and Slack upload receipt recovery.

**Subsystem affected**

`server/` — chat delivery and background work scheduling.

**Current behavior**

The app's one-second chat timer scans the three queues together with
provider and action maintenance. Empty queues continue to query
PostgreSQL.

**Proposed behavior**

Register three independent workers with the existing delivery
coordinator. Producers announce work inside their database transactions.
Commit wakes the corresponding worker. Startup restores outstanding
work; persisted retries and stale claims supply the next deadline. An
empty queue has no recurring scan.

**Reason and benefit**

Remove avoidable queue reads without requiring a new table, an external
scheduler, or a persistent PostgreSQL listener.

**Breaking changes**

No schema or public API change. Work notifications are scoped to the
app's shared database owner. Direct SQL and independent database roots
do not send these notifications; startup still scans durable work.

**Additional context**

Refs #15625, #15750, and #15757 for the preceding sleep work. Related:
#15153 reduces chat query allocation while retaining polling. This PR
changes when the three queue workers run; it does not add query caching.

## What Changed

- Register independent incoming delivery, publication, and Slack receipt
workers on the shared scheduler.
- Add transaction work signals at queue writes, endpoint state changes,
and action transitions that unblock publications. Keep provider I/O
outside the added transactions.
- Restore deadlines from durable retry times, abandoned claims, and
publication lease expiry. Share publication and receipt eligibility
rules between dispatch and deadline queries. Respect incoming
conversation order and lifecycle dependencies.
- Exclude publications blocked by unresolved Slack tool sends. Wake them
when the tool receipt changes.
- Wake publication dispatch when an endpoint send finishes so free slots
refill. Preserve per-endpoint concurrency, conversation order, and
conservative handling of uncertain sends.
- Wake receipt recovery when a stale publication becomes
`delivery_unknown`, and wake queues when a provider connection recovers.
- Commit interaction publications with their callback tokens so dispatch
cannot see an incomplete card.
- Keep provider runtimes, other action outboxes, milestone and
wake-notice projection, Teams protocol maintenance, GitHub recovery, and
Slack session synchronization on their existing maintenance cadence.
This PR does not make the whole chat subsystem idle.
- Update the instance sleep inventory with this partial migration and
its remaining work.

## Verification

- All 1,225 PostgreSQL chat integration cases passed across the three
hosted chat shards. These cover retry and lease deadlines, blocked
followers, unresolved Slack sends, provider recovery, commit visibility,
and endpoint slot refill.
- 32 local app-wiring, queue-worker, and delivery-coordinator tests
passed after updating the wiring assertion for event-driven dispatch.
Fake timers cover an hour of idle time without queue scans.
- Related publication, interaction, receipt, and wake-notice tests
passed locally. The 13 interaction integration cases include atomic
rollback when callback-token storage fails.
- Local `pnpm typecheck`, `pnpm build`, and final server typecheck
passed. All hosted checks passed on the final commit, including build,
typecheck, test shards, release registry checks, and the canary dry run
(54 successful checks; two Storybook checks skipped for this backend
change).
- The last local PostgreSQL rerun was skipped because the embedded
database startup probe failed after five attempts. The hosted PostgreSQL
cases ran successfully; their skips only partition cases across shards.
- `git diff --check` passed. Greptile is 5/5 on the final commit with no
unresolved findings.

## Risks

- Every producer that makes queued work eligible must send a transaction
work signal. The tests exercise commit visibility, concurrent wakes,
worker failures, and restart recovery. Future direct SQL writers must
participate in this contract.
- The existing database work-signal layer reconciles uncertain
transaction outcomes. This PR adds no work-signal table or schema
migration.
- Provider and action maintenance still polls. Further changes are
required before an idle chat-enabled instance can sleep.
- Tests use real PostgreSQL with simulated provider responses; they do
not send messages through live provider accounts.

## Model Used

OpenAI Codex, based on GPT-6, with reasoning, tool use, and code
execution. The exact serving model ID and context-window size are not
exposed in this session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [ ] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-10 09:13:44 -05:00

1283 lines
41 KiB
TypeScript

import { randomUUID } from "node:crypto";
import { and, eq, sql } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
agents,
chatActions,
chatConversations,
chatEndpoints,
chatPublications,
companies,
createDb,
documentRevisions,
documents,
heartbeatRuns,
issueComments,
issueDocuments,
issueThreadInteractions,
issues,
toolApplications,
toolConnections,
} from "@paperclipai/db";
import type { ChatProvider } from "@paperclipai/shared";
import { enqueueIssueInteractionChatPublications, enqueueTerminalIssueInteractionChatPublications } from "../services/chat-interaction-publications.js";
import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported
? describe
: describe.skip;
describeEmbeddedPostgres(
"terminal native chat interaction publications",
() => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<
ReturnType<typeof startEmbeddedPostgresTestDatabase>
> | null = null;
const previousPublicUrl = process.env.PAPERCLIP_PUBLIC_URL;
beforeAll(async () => {
process.env.PAPERCLIP_PUBLIC_URL = "https://paperclip.example";
tempDb = await startEmbeddedPostgresTestDatabase(
"paperclip-terminal-chat-interaction-",
);
db = createDb(tempDb.connectionString);
}, 20_000);
afterAll(async () => {
if (previousPublicUrl === undefined)
delete process.env.PAPERCLIP_PUBLIC_URL;
else process.env.PAPERCLIP_PUBLIC_URL = previousPublicUrl;
await tempDb?.cleanup();
});
async function seedBoundIssue(
providers: ChatProvider[] = ["slack", "telegram"],
) {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: `Terminal chat ${companyId}`,
issuePrefix: `TC${companyId.replaceAll("-", "").slice(0, 6).toUpperCase()}`,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "Native chat agent",
role: "operator",
status: "idle",
adapterType: "paperclip_runner",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Resolve the native provider card",
status: "in_progress",
priority: "medium",
assigneeAgentId: agentId,
});
const endpointIds: Record<string, string> = {};
for (const provider of providers) {
const applicationId = randomUUID();
const connectionId = randomUUID();
const endpointId = randomUUID();
endpointIds[provider] = endpointId;
await db.insert(toolApplications).values({
id: applicationId,
companyId,
applicationKey: `chat:${provider}:${endpointId}`,
name: `${provider} ${endpointId}`,
type: "chat",
status: "active",
});
await db.insert(toolConnections).values({
id: connectionId,
companyId,
applicationId,
name: `${provider} channel`,
uid: `chat-${provider}-${endpointId}`,
connectionPurpose: "channel",
transport: "chat_sdk",
status: "active",
enabled: true,
});
await db.insert(chatEndpoints).values({
id: endpointId,
companyId,
connectionId,
provider,
publicId: randomUUID(),
assignedAgentId: agentId,
status: "active",
capabilities: {
threads: true,
directMessages: true,
nativeStreaming: false,
messageEdits: true,
messageDeletes: false,
reactions: true,
files: true,
cards: true,
actions: true,
modals: provider !== "telegram",
slashCommands: true,
ephemeralMessages: provider === "slack",
proactiveDirectMessages: true,
},
});
await db.insert(chatConversations).values({
companyId,
endpointId,
issueId,
externalConversationId: `${provider}-conversation`,
externalThreadId: `${provider}:thread:${issueId}`,
externalLabel: `${provider} thread`,
state: "active",
});
}
return { agentId, companyId, endpointIds, issueId };
}
async function markInteractionCardsPublished(
companyId: string,
interactionId: string,
) {
const originals = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
interactionId,
),
),
);
for (const original of originals) {
await db
.update(chatPublications)
.set({
state: "published",
providerMessageId: `provider-${original.endpointId}`,
})
.where(eq(chatPublications.id, original.id));
}
return originals;
}
async function publicationsForInteraction(
companyId: string,
interactionId: string,
) {
return db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
interactionId,
),
),
);
}
function interceptAfterFirstSelect(
afterSelect: () => Promise<void>,
): Parameters<typeof enqueueTerminalIssueInteractionChatPublications>[0] {
let intercepted = false;
const wrapQuery = (query: any): any =>
new Proxy(query, {
get(target, property) {
if (property === "then") {
return (onFulfilled: unknown, onRejected: unknown) =>
Promise.resolve(target)
.then(async (value) => {
if (!intercepted) {
intercepted = true;
await afterSelect();
}
return value;
})
.then(onFulfilled as never, onRejected as never);
}
const member = Reflect.get(target, property, target);
return typeof member === "function"
? (...args: unknown[]) =>
wrapQuery(Reflect.apply(member, target, args))
: member;
},
});
return {
select: ((...args: unknown[]) =>
wrapQuery(Reflect.apply(db.select, db, args))) as typeof db.select,
insert: db.insert.bind(db),
update: db.update.bind(db),
transaction: db.transaction.bind(db),
};
}
it("publishes prompts only for the endpoint's verified immutable agent", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const foreignAgentId = randomUUID();
await db.insert(agents).values({
id: foreignAgentId,
companyId: fixture.companyId,
name: "Foreign agent",
role: "operator",
status: "idle",
adapterType: "paperclip_runner",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
const foreignRunId = randomUUID();
await db.insert(heartbeatRuns).values({
id: foreignRunId,
companyId: fixture.companyId,
agentId: foreignAgentId,
status: "succeeded",
});
const question = {
kind: "ask_user_questions" as const,
payload: {
version: 1 as const,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single" as const,
allowOther: false,
options: [{ id: "high", label: "High" }],
},
],
},
};
const userAuthored = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
question,
{ userId: "board-user" },
);
const foreignAuthored = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
question,
{ agentId: foreignAgentId },
);
const mismatchedRun = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{ ...question, sourceRunId: foreignRunId },
{ agentId: foreignAgentId },
);
await expect(
publicationsForInteraction(fixture.companyId, userAuthored.id),
).resolves.toEqual([]);
await expect(
publicationsForInteraction(fixture.companyId, foreignAuthored.id),
).resolves.toEqual([]);
await expect(
publicationsForInteraction(fixture.companyId, mismatchedRun.id),
).resolves.toEqual([]);
const assignedAgent = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
question,
{ agentId: fixture.agentId },
);
const publications = await publicationsForInteraction(
fixture.companyId,
assignedAgent.id,
);
expect(publications).toHaveLength(2);
expect(
publications.every(
(row) =>
row.idempotencyKey ===
`interaction:${assignedAgent.id}:${row.endpointId}`,
),
).toBe(true);
});
it("rolls back an interaction publication if its callback tokens cannot be stored", async () => {
const fixture = await seedBoundIssue(["telegram"]);
const interaction = await issueThreadInteractionService(db).create(
{ id: fixture.issueId, companyId: fixture.companyId },
{ kind: "request_confirmation", continuationPolicy: "wake_assignee", payload: { version: 1, prompt: "Proceed?" } },
{ agentId: fixture.agentId },
);
expect(await publicationsForInteraction(fixture.companyId, interaction.id)).toHaveLength(1);
await db.delete(chatActions).where(eq(chatActions.companyId, fixture.companyId));
await db.delete(chatPublications).where(eq(chatPublications.companyId, fixture.companyId));
const failTokens = (insert: typeof db.insert): typeof db.insert =>
((table: Parameters<typeof db.insert>[0]) => {
if (table === chatActions) throw new Error("Token write failed");
return insert(table);
}) as typeof db.insert;
const failingDb: Parameters<typeof enqueueIssueInteractionChatPublications>[0] = {
select: db.select.bind(db),
update: db.update.bind(db),
insert: failTokens(db.insert.bind(db)),
transaction: callback => db.transaction(async tx => {
tx.insert = failTokens(tx.insert.bind(tx));
return callback(tx);
}),
};
await expect(enqueueIssueInteractionChatPublications(failingDb, interaction))
.rejects.toThrow("Token write failed");
expect(await publicationsForInteraction(fixture.companyId, interaction.id)).toEqual([]);
});
it("keeps unsupported governance interactions authoritative in Paperclip", async () => {
const fixture = await seedBoundIssue();
const interaction = await issueThreadInteractionService(db).create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "request_checkbox_confirmation",
payload: {
version: 1,
prompt: "Choose deployment regions",
options: [{ id: "us", label: "US" }],
},
},
{ agentId: fixture.agentId },
);
await expect(
publicationsForInteraction(fixture.companyId, interaction.id),
).resolves.toEqual([]);
});
it("settles a card when the dispatcher claims it between the read and cancellation CAS", async () => {
const fixture = await seedBoundIssue(["slack"]);
const service = issueThreadInteractionService(db);
const interaction = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "ask_user_questions",
continuationPolicy: "wake_assignee",
payload: {
version: 1,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single",
allowOther: false,
options: [{ id: "high", label: "High" }],
},
],
},
},
{ agentId: fixture.agentId },
);
const [original] = await publicationsForInteraction(
fixture.companyId,
interaction.id,
);
if (!original) throw new Error("Expected an original publication");
await db
.update(issueThreadInteractions)
.set({
status: "answered",
result: {
version: 1,
answers: [{ questionId: "priority", optionIds: ["high"] }],
summaryMarkdown: null,
},
resolvedByUserId: "board-user",
resolvedAt: new Date(),
updatedAt: new Date(),
})
.where(eq(issueThreadInteractions.id, interaction.id));
const answered = await service.getById(interaction.id);
if (!answered) throw new Error("Expected an answered interaction");
const raceDb = interceptAfterFirstSelect(async () => {
const claimed = await db
.update(chatPublications)
.set({ state: "streaming", updatedAt: new Date() })
.where(
and(
eq(chatPublications.id, original.id),
eq(chatPublications.state, "pending"),
),
)
.returning({ id: chatPublications.id });
expect(claimed).toEqual([{ id: original.id }]);
});
await enqueueTerminalIssueInteractionChatPublications(raceDb, answered);
await expect(
db
.select({ state: chatPublications.state })
.from(chatPublications)
.where(eq(chatPublications.id, original.id)),
).resolves.toEqual([{ state: "streaming" }]);
const publications = await publicationsForInteraction(
fixture.companyId,
interaction.id,
);
expect(publications).toEqual(
expect.arrayContaining([
expect.objectContaining({
idempotencyKey: `interaction-resolution:${interaction.id}:${fixture.endpointIds.slack}`,
state: "pending",
payload: expect.objectContaining({ text: "Answered: High." }),
}),
]),
);
});
it("settles an ambiguously delivered link-only card after an operator marks it delivered", async () => {
const fixture = await seedBoundIssue(["github"]);
const service = issueThreadInteractionService(db);
const interaction = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "request_confirmation",
continuationPolicy: "wake_assignee",
payload: { version: 1, prompt: "Ship the GitHub release?" },
},
{ agentId: fixture.agentId },
);
const [original] = await publicationsForInteraction(
fixture.companyId,
interaction.id,
);
if (!original) throw new Error("Expected a GitHub link-only card");
await db
.update(chatPublications)
.set({ state: "published", providerMessageId: null })
.where(eq(chatPublications.id, original.id));
await service.acceptInteraction(
{
id: fixture.issueId,
companyId: fixture.companyId,
projectId: null,
goalId: null,
},
interaction.id,
{},
{ userId: "board-user" },
);
await expect(
db
.select({
idempotencyKey: chatPublications.idempotencyKey,
payload: chatPublications.payload,
state: chatPublications.state,
})
.from(chatPublications)
.where(
eq(
chatPublications.idempotencyKey,
`interaction-resolution:${interaction.id}:${fixture.endpointIds.github}`,
),
),
).resolves.toEqual([
{
idempotencyKey: `interaction-resolution:${interaction.id}:${fixture.endpointIds.github}`,
payload: expect.objectContaining({
interactionId: interaction.id,
text: "Accepted.",
}),
state: "pending",
},
]);
});
it("settles form answers without claiming a surface or exposing free text", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const interaction = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "ask_user_questions",
payload: {
version: 1,
questions: [
{
id: "choice",
prompt: "Choose a tree",
selectionMode: "single",
allowOther: false,
options: [{ id: "cedar", label: "Cedar" }],
},
{
id: "label",
prompt: "Enter a label",
selectionMode: "single",
allowOther: true,
options: [
{ id: "text", label: "Write a label", freeText: true },
],
},
],
},
},
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(fixture.companyId, interaction.id);
await service.answerQuestions(
{ id: fixture.issueId, companyId: fixture.companyId },
interaction.id,
{
answers: [
{ questionId: "choice", optionIds: ["cedar"] },
{
questionId: "label",
optionIds: [],
otherText: "private free-text label",
},
],
},
{ userId: "board-user" },
);
const settlements = (
await publicationsForInteraction(fixture.companyId, interaction.id)
).filter((row) => row.state === "pending");
expect(settlements).toHaveLength(2);
for (const settlement of settlements) {
expect(settlement.payload).toMatchObject({
text: "Answered.",
card: { kind: "question", body: "Answered." },
});
expect(settlement.payload.card?.actions ?? []).toHaveLength(0);
expect(JSON.stringify(settlement.payload)).not.toContain(
"private free-text label",
);
expect(JSON.stringify(settlement.payload)).not.toContain(
"Answered in Paperclip",
);
}
});
it("settles UI answers and skips on every provider-native card with no live actions", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const questionInput = {
kind: "ask_user_questions" as const,
payload: {
version: 1 as const,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single" as const,
allowOther: false,
options: [
{ id: "high", label: "High" },
{ id: "normal", label: "Normal" },
],
},
],
},
};
const answeredQuestion = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
questionInput,
{ agentId: fixture.agentId },
);
expect(
await markInteractionCardsPublished(
fixture.companyId,
answeredQuestion.id,
),
).toHaveLength(2);
const answered = await service.answerQuestions(
{ id: fixture.issueId, companyId: fixture.companyId },
answeredQuestion.id,
{ answers: [{ questionId: "priority", optionIds: ["high"] }] },
{ userId: "board-user" },
);
const answeredSettlements = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
answeredQuestion.id,
),
eq(chatPublications.state, "pending"),
),
);
expect(answeredSettlements).toHaveLength(2);
expect(answeredSettlements).toEqual(
expect.arrayContaining(
Object.values(fixture.endpointIds).map((endpointId) =>
expect.objectContaining({
endpointId,
idempotencyKey: `interaction-resolution:${answeredQuestion.id}:${endpointId}`,
payload: expect.objectContaining({
text: "Answered: High.",
card: expect.objectContaining({
kind: "question",
title: "Which priority?",
body: "Answered: High.",
}),
}),
}),
),
),
);
expect(
answeredSettlements.every(
(row) => (row.payload.card?.actions ?? []).length === 0,
),
).toBe(true);
await expect(
enqueueTerminalIssueInteractionChatPublications(db, answered),
).resolves.toEqual([]);
await expect(
db
.select({ status: chatActions.status })
.from(chatActions)
.where(
eq(
sql<string>`${chatActions.payload}->>'interactionId'`,
answeredQuestion.id,
),
),
).resolves.toEqual([
{ status: "expired" },
{ status: "expired" },
{ status: "expired" },
{ status: "expired" },
]);
const skippedQuestion = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
questionInput,
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(
fixture.companyId,
skippedQuestion.id,
);
await service.skipInteraction(
{
id: fixture.issueId,
companyId: fixture.companyId,
status: "in_progress",
},
skippedQuestion.id,
{},
{ userId: "board-user" },
);
const skippedSettlements = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
skippedQuestion.id,
),
eq(chatPublications.state, "pending"),
),
);
expect(skippedSettlements).toHaveLength(2);
expect(
skippedSettlements.every(
(row) => row.payload.card?.body === "Skipped in Paperclip.",
),
).toBe(true);
expect(
skippedSettlements.every(
(row) => (row.payload.card?.actions ?? []).length === 0,
),
).toBe(true);
await expect(
db
.select({ id: chatActions.id })
.from(chatActions)
.where(
and(
eq(chatActions.kind, "interaction_wakeup"),
eq(
sql<string>`${chatActions.payload}->>'interactionId'`,
skippedQuestion.id,
),
),
),
).resolves.toEqual([]);
});
it("settles native and link-only confirmation cards without inventing controls", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
for (const outcome of ["accept", "reject"] as const) {
const interaction = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "request_confirmation",
continuationPolicy: "wake_assignee",
payload: { version: 1, prompt: `Should we ${outcome}?` },
},
{ agentId: fixture.agentId },
);
const originals = await markInteractionCardsPublished(
fixture.companyId,
interaction.id,
);
expect(originals).toHaveLength(2);
expect(
originals.find((row) => row.endpointId === fixture.endpointIds.slack)
?.payload.card?.actions,
).toEqual([
expect.objectContaining({ type: "link", label: "Open in Paperclip" }),
]);
expect(
originals.find(
(row) => row.endpointId === fixture.endpointIds.telegram,
)?.payload.card?.actions,
).toEqual([
expect.objectContaining({ type: "callback" }),
expect.objectContaining({ type: "callback" }),
]);
if (outcome === "accept") {
await service.acceptInteraction(
{
id: fixture.issueId,
companyId: fixture.companyId,
projectId: null,
goalId: null,
},
interaction.id,
{},
{ userId: "board-user" },
);
} else {
await service.rejectInteraction(
{ id: fixture.issueId, companyId: fixture.companyId },
interaction.id,
{},
{ userId: "board-user" },
);
}
const settlements = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
interaction.id,
),
eq(chatPublications.state, "pending"),
),
);
expect(settlements).toHaveLength(2);
expect(settlements).toEqual(
expect.arrayContaining(
Object.values(fixture.endpointIds).map((endpointId) =>
expect.objectContaining({
endpointId,
idempotencyKey: `interaction-resolution:${interaction.id}:${endpointId}`,
payload: expect.objectContaining({
text: outcome === "accept" ? "Accepted." : "Rejected.",
card: expect.objectContaining({
kind: "confirmation",
title: `Should we ${outcome}?`,
body: outcome === "accept" ? "Accepted" : "Rejected",
}),
}),
}),
),
),
);
expect(
settlements.every(
(settlement) =>
(settlement.payload.card?.actions ?? []).length === 0,
),
).toBe(true);
await expect(
db
.select({
kind: chatActions.kind,
payload: chatActions.payload,
status: chatActions.status,
})
.from(chatActions)
.where(
and(
eq(chatActions.kind, "interaction_wakeup"),
eq(
sql<string>`${chatActions.payload}->>'interactionId'`,
interaction.id,
),
),
),
).resolves.toEqual([
{
kind: "interaction_wakeup",
payload: expect.objectContaining({
agentId: fixture.agentId,
interactionId: interaction.id,
interactionStatus: outcome === "accept" ? "accepted" : "rejected",
issueId: fixture.issueId,
requestedByActorId: "board-user",
requestedByActorType: "user",
}),
status: "issued",
},
]);
}
});
it("expires both modal action tokens when a form-backed question is answered", async () => {
const fixture = await seedBoundIssue(["slack"]);
const service = issueThreadInteractionService(db);
const interaction = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "ask_user_questions",
continuationPolicy: "wake_assignee",
payload: {
version: 1,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single",
allowOther: false,
options: [{ id: "high", label: "High" }],
},
{
id: "owner",
prompt: "Who owns it?",
selectionMode: "single",
allowOther: false,
options: [{ id: "agent", label: "Agent" }],
},
],
},
},
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(fixture.companyId, interaction.id);
await service.answerQuestions(
{ id: fixture.issueId, companyId: fixture.companyId },
interaction.id,
{
answers: [
{ questionId: "priority", optionIds: ["high"] },
{ questionId: "owner", optionIds: ["agent"] },
],
},
{ userId: "board-user" },
);
const actions = await db
.select({ kind: chatActions.kind, status: chatActions.status })
.from(chatActions)
.where(
eq(
sql<string>`${chatActions.payload}->>'interactionId'`,
interaction.id,
),
);
expect(actions).toEqual(
expect.arrayContaining([
{ kind: "question_form_open", status: "expired" },
{ kind: "question_form_submit", status: "expired" },
]),
);
expect(actions).toHaveLength(2);
});
it("cancels unsent native and link-only cards instead of publishing stale prompts", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const interaction = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "request_confirmation",
payload: { version: 1, prompt: "Proceed?" },
},
{ agentId: fixture.agentId },
);
await service.acceptInteraction(
{
id: fixture.issueId,
companyId: fixture.companyId,
projectId: null,
goalId: null,
},
interaction.id,
{},
{ userId: "board-user" },
);
const publications = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
interaction.id,
),
),
);
expect(publications).toHaveLength(2);
expect(publications).toEqual(
expect.arrayContaining(
Object.values(fixture.endpointIds).map((endpointId) =>
expect.objectContaining({
endpointId,
idempotencyKey: `interaction:${interaction.id}:${endpointId}`,
state: "cancelled",
redactedError:
"Interaction was resolved before provider publication",
}),
),
),
);
expect(
publications.some((row) =>
row.payload.card?.actions?.some((action) => action.type === "link"),
),
).toBe(true);
});
it("settles visible question cards when a newer interaction supersedes them", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const input = {
kind: "ask_user_questions" as const,
payload: {
version: 1 as const,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single" as const,
allowOther: false,
options: [{ id: "high", label: "High" }],
},
],
},
};
const original = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
input,
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(fixture.companyId, original.id);
const replacement = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
input,
{ agentId: fixture.agentId },
);
await expect(service.getById(original.id)).resolves.toMatchObject({
status: "expired",
result: {
expirationReason: "superseded_by_newer_interaction",
supersededByInteractionId: replacement.id,
},
});
const publications = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
original.id,
),
),
);
const settlements = publications.filter((row) =>
row.idempotencyKey.startsWith("interaction-resolution:"),
);
expect(settlements).toHaveLength(2);
expect(
settlements.every(
(row) =>
row.state === "pending" &&
row.payload.text === "Expired: replaced by a newer request." &&
row.payload.card?.body === "Expired: replaced by a newer request" &&
(row.payload.card.actions ?? []).length === 0,
),
).toBe(true);
});
it("settles live and historical human-comment supersessions", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const confirmation = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "request_confirmation",
payload: {
version: 1,
prompt: "Proceed with the release?",
supersedeOnUserComment: true,
},
},
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(fixture.companyId, confirmation.id);
const liveCommentId = randomUUID();
await service.expireRequestConfirmationsSupersededByComment(
{ id: fixture.issueId, companyId: fixture.companyId },
{
id: liveCommentId,
authorUserId: "board-user",
createdByRunId: null,
createdAt: new Date(Date.now() + 1_000),
},
{ userId: "board-user" },
);
const liveSettlements = await db
.select()
.from(chatPublications)
.where(
eq(
chatPublications.idempotencyKey,
`interaction-resolution:${confirmation.id}:${fixture.endpointIds.telegram}`,
),
);
expect(liveSettlements).toEqual([
expect.objectContaining({
payload: expect.objectContaining({
text: "Expired: superseded by a newer reply.",
card: expect.objectContaining({
body: "Expired: superseded by a newer reply",
}),
}),
}),
]);
expect(liveSettlements[0]?.payload.card?.actions ?? []).toHaveLength(0);
const question = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "ask_user_questions",
payload: {
version: 1,
supersedeOnUserComment: true,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single",
allowOther: false,
options: [{ id: "high", label: "High" }],
},
],
},
},
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(fixture.companyId, question.id);
const questionCreatedAt = new Date("2026-09-05T12:00:00.000Z");
await db
.update(issueThreadInteractions)
.set({ createdAt: questionCreatedAt, updatedAt: questionCreatedAt })
.where(eq(issueThreadInteractions.id, question.id));
await db.insert(issueComments).values({
id: randomUUID(),
companyId: fixture.companyId,
issueId: fixture.issueId,
authorUserId: "board-user",
authorType: "user",
body: "Use the high-priority path.",
createdAt: new Date("2026-09-05T12:01:00.000Z"),
updatedAt: new Date("2026-09-05T12:01:00.000Z"),
});
await service.expireRequestConfirmationsSupersededByHistoricalComments({
id: fixture.issueId,
companyId: fixture.companyId,
});
const historicalSettlements = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
question.id,
),
eq(chatPublications.state, "pending"),
),
);
expect(historicalSettlements).toHaveLength(2);
expect(
historicalSettlements.every(
(row) =>
row.payload.card?.body === "Expired: superseded by a newer reply" &&
(row.payload.card.actions ?? []).length === 0,
),
).toBe(true);
});
it("cancels stale-target prompts and settles terminal-issue expirations", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
const documentId = randomUUID();
const revisionId = randomUUID();
const nextRevisionId = randomUUID();
await db.insert(documents).values({
id: documentId,
companyId: fixture.companyId,
title: "Plan",
format: "markdown",
latestBody: "v1",
latestRevisionId: revisionId,
latestRevisionNumber: 1,
});
await db.insert(issueDocuments).values({
companyId: fixture.companyId,
issueId: fixture.issueId,
documentId,
key: "plan",
});
await db.insert(documentRevisions).values({
id: revisionId,
companyId: fixture.companyId,
documentId,
revisionNumber: 1,
title: "Plan",
format: "markdown",
body: "v1",
});
const confirmation = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "request_confirmation",
payload: {
version: 1,
prompt: "Apply this plan?",
target: {
type: "issue_document",
issueId: fixture.issueId,
documentId,
key: "plan",
revisionId,
revisionNumber: 1,
},
},
},
{ agentId: fixture.agentId },
);
await service.expireStaleRequestConfirmationsForIssueDocument(
{ id: fixture.issueId, companyId: fixture.companyId },
{
id: documentId,
key: "plan",
latestRevisionId: nextRevisionId,
latestRevisionNumber: 2,
},
{ userId: "board-user" },
);
const stalePublications = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
confirmation.id,
),
),
);
expect(stalePublications).toHaveLength(2);
expect(
stalePublications.every(
(row) =>
row.state === "cancelled" &&
row.redactedError ===
"Interaction was resolved before provider publication",
),
).toBe(true);
const question = await service.create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "ask_user_questions",
payload: {
version: 1,
questions: [
{
id: "priority",
prompt: "Which priority?",
selectionMode: "single",
allowOther: false,
options: [{ id: "high", label: "High" }],
},
],
},
},
{ agentId: fixture.agentId },
);
await markInteractionCardsPublished(fixture.companyId, question.id);
await service.expirePendingInteractionsForTerminalIssue(
{
id: fixture.issueId,
companyId: fixture.companyId,
status: "done",
},
{ userId: "board-user" },
);
const closedSettlements = await db
.select()
.from(chatPublications)
.where(
and(
eq(chatPublications.companyId, fixture.companyId),
eq(
sql<string>`${chatPublications.payload}->>'interactionId'`,
question.id,
),
eq(chatPublications.state, "pending"),
),
);
expect(closedSettlements).toHaveLength(2);
expect(
closedSettlements.every(
(row) =>
row.payload.text === "Expired: task is closed." &&
row.payload.card?.body === "Expired: task is closed" &&
(row.payload.card.actions ?? []).length === 0,
),
).toBe(true);
});
},
);