Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-10-25 02:26:12 +02:00
Code Issues Packages Projects Releases Wiki Activity

perf(chat): wake delivery queues from commits and deadlines (#15802)

Browse Source 
## 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>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-10 09:13:44 -05:00
1 parent 1fb3f9746e
commit 19e9facc5b
20 files changed
+1852 -1104
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md
+16 -5
View File
@@ -415,12 +415,23 @@ from event-driven normal paths and route only real deadlines through the interna
scheduler. Preserve durable delivery, startup recovery, and domain-level claims.
Longer polling intervals alone do not complete an item.
- [ ] **JOB-01 — Chat reconciliation.** Current: every **1s**, with multiple
runtime/delivery/publication lanes and SQL even without connectors. Change:
disarm empty lanes and activate on committed events or deadlines. **Complete
when:** a tenant with no chat obligations issues no chat SQL and new inbound,
publication, and recovery work still runs. Sources:
- [ ] **JOB-01 — Chat reconciliation.** Partially migrated on 2026-10-10.
Inbound deliveries, outbound publications, and Slack upload receipt recovery
now use transaction-aware commit wakes and durable retry/claim deadlines.
Empty queues disarm after startup. Publication deadlines use the dispatch
eligibility predicate so paused endpoints, unknown outcomes, blocked followers,
and receipts owned by a live sender do not cause empty scans. Completion wakes
refill free endpoint slots without waiting for slow peers. No new tables.
Remaining **1s** maintenance: provider runtimes, provider-specific action
outboxes, run milestone/wake-notice projection, GitHub operation recovery,
Teams protocol maintenance, and Slack session sync. GitHub pre-ingress recovery
must still inspect provider history: a lost callback has no local queue row.
This slice does **not** yet make the whole chat subsystem quiet. Independent
DB roots/direct SQL require startup recovery, as with the other migrated queues.
**Complete when:** all remaining lanes are demand/deadline-driven and a tenant
with no chat obligations issues no chat SQL. Sources:
[app.ts](../../server/src/app.ts),
[chat-delivery-work.ts](../../server/src/services/chat-delivery-work.ts),
[chat-channels.ts](../../server/src/services/chat-channels.ts).
- [x] **JOB-02 — Email reconciliation.** Replaced the unconditional **1s**
server/src/__tests__/app-private-hostname-gate.test.ts
+10 -9
View File
@@ -90,17 +90,18 @@ describe("createChatReconciliationCoordinator", () => {
expect(ordinary).not.toContain("processFailedGitHubWebhookDeliveries()");
});
it("wires periodic publication reconciliation to bounded scheduled refill rather than awaiting provider sends", () => {
it("wires bounded publication refill to commit workers and isolates periodic maintenance", () => {
const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
const flush = source.slice(
source.indexOf("const flushChatPublications ="),
source.indexOf("const flushChatPublications =") + 600,
expect(source).toContain("registerChatDeliveryWork(deliveryWork, chatChannels,");
const maintenance = source.slice(
source.indexOf("const reconcileChatPublicationMaintenance ="),
source.indexOf("const chatReconciliation ="),
);
expect(flush).toContain("await chatChannels.schedulePendingPublications()");
expect(flush).not.toContain("chatChannels.processPendingPublications()");
expect(
flush.slice(0, flush.indexOf("const chatReconciliation")),
).not.toContain("await enqueueChatRunMilestones");
expect(maintenance).toContain("await chatChannels.processPublicationMaintenance()");
expect(maintenance).not.toContain("chatChannels.schedulePendingPublications()");
expect(maintenance).not.toContain("chatChannels.processPendingPublications()");
const workers = readFileSync(new URL("../services/chat-delivery-work.ts", import.meta.url), "utf8");
expect(workers).toContain("run: service.scheduleQueuedPublications");
// The service integration tests hold real publication workers while this
// scheduled method returns; app shutdown must also join those workers.
expect(source).toContain("await chatChannels.shutdown()");
server/src/__tests__/chat-channels.integration.test.ts
+183 -5
View File
@@ -1,3 +1,7 @@
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
import { notifyChatPublicationWork } from "../services/chat-work-notifications.js";
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
import { chatSlackRegistrations, toolOauthStates } from "@paperclipai/db";
import { buildSlackAppManifest } from "@paperclipai/shared";
import { toolAccessService } from "../services/tool-access.js";
@@ -7345,7 +7349,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
const countBefore = postFetch.mock.calls.length;
await executeConnectorTool(db, binding, "slack_post_message", uncertain);
expect(postFetch.mock.calls).toHaveLength(countBefore);
await expect(executeSlackTool(db, binding, "slack_delivery", { actionId: uncertainAction.id }, postFetch as typeof fetch)).resolves.toMatchObject({ state: "delivered", receipt: { reconciled: true } });
const receiptSettled = vi.fn();
const unsubscribeReceipt = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatPublications, receiptSettled);
try {
await expect(executeSlackTool(db, binding, "slack_delivery", { actionId: uncertainAction.id }, postFetch as typeof fetch)).resolves.toMatchObject({ state: "delivered", receipt: { reconciled: true } });
expect(receiptSettled).toHaveBeenCalledTimes(1);
} finally { unsubscribeReceipt(); }
await expect(executeConnectorTool(db, binding, "slack_create_channel", { name: "approval-required", is_private: true, idempotencyKey: randomUUID() })).rejects.toMatchObject({ reasonCode: expect.stringMatching(/approval/) });
expect(postFetch.mock.calls.some(call => String(call[0]).endsWith("/conversations.create"))).toBe(false);
const [approval] = await db.select().from(toolActionRequests).where(eq(toolActionRequests.companyId, fixture.companyId));
@@ -14482,6 +14491,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
.onDiscordGatewayEvent;
if (!onGatewayEvent) throw new Error("Expected Discord Gateway callback");
const recovered = vi.fn();
const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatPublications, recovered);
const disconnect = Promise.resolve(
onGatewayEvent({
endpointId: endpoint.id,
@@ -14505,7 +14516,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
await Promise.resolve();
expect(recoveryFinished).toBe(false);
releaseDisconnect();
await Promise.all([disconnect, recovery]);
try {
await Promise.all([disconnect, recovery]);
expect(recovered).toHaveBeenCalled();
} finally { unsubscribe(); }
await expect(context.service.get(endpoint.id)).resolves.toMatchObject({
status: "active",
healthMessage: "Connected",
@@ -17360,10 +17374,22 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
const fixture = await seedCompany();
const { endpoint, service } = await configuredSlackEndpoint(fixture);
const configuredEndpoint = await service.get(endpoint.id);
const originalTransaction = db.transaction.bind(db);
const transaction = vi.spyOn(db, "transaction");
transaction.mockRejectedValueOnce(
new Error("injected lifecycle persistence failure"),
);
let injected = false;
transaction.mockImplementation(async (callback, options) => {
const [delivery] = await db.select({ state: chatDeliveries.state }).from(chatDeliveries).where(and(
eq(chatDeliveries.endpointId, endpoint.id),
eq(chatDeliveries.providerEventId, "lifecycle:Ev-lifecycle-retry"),
));
// Fail the lifecycle effect after durable admission and claim, regardless
// of how many transaction boundaries those writes require.
if (!injected && delivery?.state === "processing") {
injected = true;
throw new Error("injected lifecycle persistence failure");
}
return originalTransaction(callback, options);
});
await expect(
service.handleWebhook(
endpoint.publicId,
@@ -32754,6 +32780,158 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
return { ...fixture, storage, runtime, service, lanes, enqueue, cleanup };
}
it("restores chat queue deadlines without polling blocked followers or receipt owners", async () => {
const f = await publicationLaneFixture(1);
const { service, lanes, enqueue } = f;
const lane = lanes[0]!;
try {
const head = await enqueue(0, "Head");
const tail = await enqueue(0, "Follower", 1);
const retryAt = new Date(Date.now() + 120_000);
await db.update(chatPublications).set({ state: "retry", nextAttemptAt: retryAt }).where(eq(chatPublications.id, head.id));
expect(await service.nextPublicationAt()).toBe(retryAt.getTime());
await db.update(chatPublications).set({ state: "delivery_unknown", nextAttemptAt: null }).where(eq(chatPublications.id, head.id));
expect(await service.nextPublicationAt()).toBeNull();
const leaseUntil = new Date(Date.now() + 180_000);
await db.update(chatPublications).set({ state: "streaming", updatedAt: new Date() }).where(eq(chatPublications.id, head.id));
await db.insert(chatEndpointLeases).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
leaseKey: `publication:${head.id}:0`, token: randomUUID(), expiresAt: leaseUntil });
expect(await service.nextPublicationAt()).toBe(leaseUntil.getTime());
await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
kind: "slack_file_upload_receipt", providerActionId: randomUUID(), status: "received",
payload: { publicationId: head.id, publicationAttempt: 0 } });
expect(await service.nextSlackReceiptAt()).toBeNull();
await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, lane.endpoint.id));
await db.update(chatPublications).set({ updatedAt: new Date(Date.now() - 70_000) }).where(eq(chatPublications.id, head.id));
const wake = vi.fn();
const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatReceipts, wake);
try {
await service.scheduleQueuedPublications();
expect(wake).toHaveBeenCalled();
expect(await service.nextSlackReceiptAt()).toBeLessThanOrEqual(Date.now());
await service.processPendingSlackFileUploadReceipts(); // Invalid retained payload is closed, not retried.
expect(await service.nextSlackReceiptAt()).toBeNull();
} finally { unsubscribe(); }
await db.update(chatPublications).set({ state: "cancelled" }).where(eq(chatPublications.id, tail.id));
expect(await service.nextPublicationAt()).toBeNull();
} finally { await f.cleanup(); }
});
it("restores inbound retry and abandoned wake deadlines and excludes paused queues", async () => {
const f = await publicationLaneFixture(1);
const { service, lanes } = f;
const lane = lanes[0]!;
try {
const retryAt = new Date(Date.now() + 120_000);
const [delivery] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
conversationId: lane.conversation.id, providerEventId: randomUUID(), deduplicationKey: randomUUID(),
eventKind: "message", normalizedEvent: { conversation: { externalThreadId: lane.conversation.externalThreadId } },
receivedAt: new Date(Date.now() - 1000), state: "retry", nextAttemptAt: retryAt }).returning();
const [follower] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
conversationId: lane.conversation.id, providerEventId: randomUUID(), deduplicationKey: randomUUID(),
eventKind: "message", normalizedEvent: delivery!.normalizedEvent, state: "received" }).returning();
expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
const [edit] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "message_updated", state: "received",
receivedAt: new Date(Date.now() - 2000), normalizedEvent: { ...delivery!.normalizedEvent,
message: { targetProviderEventId: delivery!.providerEventId } } }).returning();
expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
await db.update(chatDeliveries).set({ state: "filtered" }).where(eq(chatDeliveries.id, edit!.id));
expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
await db.update(chatDeliveries).set({ state: "processed" }).where(eq(chatDeliveries.id, delivery!.id));
const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
deliveryId: delivery!.id, kind: "inbound_wakeup", providerActionId: randomUUID(), status: "issued",
result: { retryAt: retryAt.toISOString() } }).returning();
expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
const claimedAt = new Date();
await db.update(chatActions).set({ status: "processing", updatedAt: claimedAt }).where(eq(chatActions.id, action!.id));
expect(await service.nextInboundDeliveryAt()).toBe(claimedAt.getTime() + 60_000);
await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, lane.endpoint.id));
expect(await service.nextInboundDeliveryAt()).toBeNull();
await db.update(chatActions).set({ status: "processed" }).where(eq(chatActions.id, action!.id));
await db.update(chatDeliveries).set({ state: "filtered" }).where(eq(chatDeliveries.id, follower!.id));
await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, lane.endpoint.id));
expect(await service.nextInboundDeliveryAt()).toBeNull();
await db.update(chatEndpoints).set({ status: "attention" }).where(eq(chatEndpoints.id, lane.endpoint.id));
await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "installation", state: "retry",
nextAttemptAt: retryAt, normalizedEvent: { lifecycle: { kind: "endpoint", provider: "slack",
providerEventId: randomUUID(), availability: "available", reason: "Recovered" } } });
expect(await service.nextInboundDeliveryAt()).toBe(retryAt.getTime());
} finally { await f.cleanup(); }
});
it("sleeps behind an unresolved Slack tool send and wakes when its receipt settles", async () => {
const f = await publicationLaneFixture(1);
const lane = f.lanes[0]!;
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
try {
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, status: "succeeded" }).returning();
const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: lane.conversation.issueId,
authorAgentId: f.assignedAgentId, createdByRunId: run!.id, body: "Already sent" }).returning();
const publication = await f.enqueue(0, "Already sent");
await db.update(chatPublications).set({ commentId: comment!.id }).where(eq(chatPublications.id, publication.id));
const [send] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id,
conversationId: lane.conversation.id, kind: "slack_tool_write", providerActionId: randomUUID(), status: "processing",
payload: { name: "slack_post_message", binding: { runId: run!.id }, args: {
channel: lane.conversation.externalConversationId.replace(/^slack:/, ""),
thread_ts: lane.conversation.externalThreadId.split(":").at(-1), text: " Already sent\n" } } }).returning();
await registerChatDeliveryWork(coordinator, f.service, () => true).ready;
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
expect(await f.service.nextPublicationAt()).toBeNull();
for (const status of ["uncertain", "processed"]) {
await db.transaction(async tx => {
await notifyChatPublicationWork(tx);
await tx.update(chatActions).set({ status }).where(eq(chatActions.id, send!.id));
});
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
}
await vi.waitFor(async () => {
const [row] = await db.select().from(chatPublications).where(eq(chatPublications.id, publication.id));
expect(row!.state).toBe("cancelled");
expect(coordinator.nextWakeAt()).toBeNull();
}, { timeout: 10_000 });
expect(lane.providerRuntime.posts).toHaveLength(0);
expect(onError).not.toHaveBeenCalled();
} finally { await coordinator.stop(); await f.cleanup(); }
});
it("wakes chat dispatch after commit and refills free endpoint slots without the maintenance loop", async () => {
const f = await publicationLaneFixture(6);
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
f.lanes[0]!.providerRuntime.postHook = () => gate;
try {
await registerChatDeliveryWork(coordinator, f.service, () => true).ready;
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 });
let settle!: () => void;
const commitGate = new Promise<void>(resolve => { settle = resolve; });
let staged!: () => void;
const stagedGate = new Promise<void>(resolve => { staged = resolve; });
const writing = db.transaction(async tx => {
await notifyChatPublicationWork(tx);
await tx.insert(chatPublications).values(f.lanes.map((lane, index) => ({ companyId: f.companyId,
endpointId: lane.endpoint.id, conversationId: lane.conversation.id, issueId: lane.conversation.issueId,
idempotencyKey: `work-test:${randomUUID()}`, payload: { text: `Queued ${index}` }, state: "pending" as const })));
staged(); await commitGate;
});
await stagedGate;
expect(f.lanes.every(lane => lane.providerRuntime.posts.length === 0)).toBe(true);
settle(); await writing;
await vi.waitFor(() => expect(f.lanes.slice(1).every(lane => lane.providerRuntime.posts.length === 1)).toBe(true), { timeout: 10_000 });
release();
await vi.waitFor(async () => {
const rows = await db.select().from(chatPublications).where(eq(chatPublications.companyId, f.companyId));
expect(rows.every(row => row.state === "published")).toBe(true);
expect(coordinator.nextWakeAt()).toBeNull();
}, { timeout: 10_000 });
expect(onError).not.toHaveBeenCalled();
} finally { release(); await coordinator.stop(); await f.cleanup(); }
});
it("refills four tracked publication endpoint lanes after a scheduled budget and joins shutdown", async () => {
const fixture = await publicationLaneFixture(6);
const { service, lanes, enqueue, storage } = fixture;
server/src/__tests__/chat-delivery-work.test.ts
+112
View File
@@ -0,0 +1,112 @@
import { afterEach, describe, expect, it, vi } from "vitest";
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { registerChatDeliveryWork } from "../services/chat-delivery-work.js";
import { notifyChatDeliveryWork, notifyChatPublicationWork } from "../services/chat-work-notifications.js";
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
afterEach(async () => {
for (const coordinator of coordinators.splice(0)) await coordinator.stop();
vi.useRealTimers();
});
function setup() {
vi.useFakeTimers();
const execute = vi.fn(async () => [{ xid: "42", status: "committed" }]);
const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>) {
return callback({ execute });
} });
const enabled = vi.fn(() => true), primary = vi.fn(() => true), onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner,
canRun: () => enabled() && primary(), canReconcile: primary, onError,
});
coordinators.push(coordinator);
const service = {
processQueuedDeliveries: vi.fn(async () => {}),
scheduleQueuedPublications: vi.fn(async () => {}),
processPendingSlackFileUploadReceipts: vi.fn(async () => {}),
nextInboundDeliveryAt: vi.fn(async (): Promise<number | null> => null),
nextPublicationAt: vi.fn(async (): Promise<number | null> => null),
nextSlackReceiptAt: vi.fn(async (): Promise<number | null> => null),
onPublicationsSettled: vi.fn(),
};
const start = () => registerChatDeliveryWork(coordinator, service, enabled);
return { owner, service, coordinator, start, enabled, primary, onError };
}
describe("chat delivery work", () => {
it("does one startup pass and no further queue scans when empty", async () => {
const s = setup();
await s.start().ready;
await vi.advanceTimersByTimeAsync(3_600_000);
for (const run of [s.service.processQueuedDeliveries, s.service.scheduleQueuedPublications, s.service.processPendingSlackFileUploadReceipts]) expect(run).toHaveBeenCalledTimes(1);
expect(s.coordinator.nextWakeAt()).toBeNull();
expect(vi.getTimerCount()).toBe(0);
});
it("coalesces commits, wakes receipt recovery after publication changes, and refills freed slots", async () => {
const s = setup();
await s.start().ready;
await s.owner.transaction(tx => notifyChatPublicationWork(tx));
await s.owner.transaction(tx => notifyChatPublicationWork(tx));
await vi.advanceTimersByTimeAsync(1);
expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(2);
expect(s.service.processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2);
expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
s.service.onPublicationsSettled.mock.calls[0]![0]();
await vi.advanceTimersByTimeAsync(1);
expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(3);
expect(vi.getTimerCount()).toBe(0);
});
it("keeps lanes independent and retains commits received during a blocked drain", async () => {
const s = setup();
let release!: () => void;
s.service.processQueuedDeliveries.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
const worker = s.start();
await vi.advanceTimersByTimeAsync(1);
await s.owner.transaction(tx => notifyChatDeliveryWork(tx));
await s.owner.transaction(tx => notifyChatPublicationWork(tx));
await vi.advanceTimersByTimeAsync(1);
expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(2);
release();
await worker.ready;
await vi.advanceTimersByTimeAsync(1);
expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(2);
});
it("restores future retries and stops when the last retry finishes", async () => {
const s = setup();
const due = Date.now() + 60_000;
s.service.nextPublicationAt.mockResolvedValue(due);
await s.start().ready;
expect(s.coordinator.nextWakeAt()).toBe(due);
await vi.advanceTimersByTimeAsync(59_999);
expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(1);
s.service.nextPublicationAt.mockResolvedValue(null);
await vi.advanceTimersByTimeAsync(1);
expect(s.service.scheduleQueuedPublications).toHaveBeenCalledTimes(2);
expect(vi.getTimerCount()).toBe(0);
});
it("does no queue SQL while standby or drain defers a first commit", async () => {
const s = setup();
s.primary.mockReturnValue(false);
await s.start().ready;
await vi.advanceTimersByTimeAsync(2000);
s.primary.mockReturnValue(true); s.enabled.mockReturnValue(false);
await s.owner.transaction(tx => notifyChatDeliveryWork(tx));
await vi.advanceTimersByTimeAsync(2000);
expect(s.service.processQueuedDeliveries).not.toHaveBeenCalled();
expect(s.service.nextInboundDeliveryAt).not.toHaveBeenCalled();
s.enabled.mockReturnValue(true);
await vi.advanceTimersByTimeAsync(1000);
expect(s.service.processQueuedDeliveries).toHaveBeenCalledTimes(1);
expect(vi.getTimerCount()).toBe(0);
});
it("retries failed selection without losing its startup recovery", async () => {
const s = setup();
s.service.nextSlackReceiptAt.mockRejectedValueOnce(new Error("connection lost"));
await s.start().ready;
expect(s.onError).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1000);
expect(s.service.processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2);
expect(vi.getTimerCount()).toBe(0);
});
});
server/src/__tests__/chat-interaction-publications.test.ts
+31 -1
View File
@@ -20,7 +20,7 @@ import {
toolConnections,
} from "@paperclipai/db";
import type { ChatProvider } from "@paperclipai/shared";
import { enqueueTerminalIssueInteractionChatPublications } from "../services/chat-interaction-publications.js";
import { enqueueIssueInteractionChatPublications, enqueueTerminalIssueInteractionChatPublications } from "../services/chat-interaction-publications.js";
import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
import {
getEmbeddedPostgresTestSupport,
@@ -226,6 +226,7 @@ describeEmbeddedPostgres(
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),
};
}
@@ -312,6 +313,35 @@ describeEmbeddedPostgres(
).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(
server/src/app.ts
+10 -8
View File
@@ -9,6 +9,7 @@ import { customerSuccessRoutes } from "./routes/customer-success.js";
import { cloudWarmStandbyMiddleware } from "./middleware/cloud-warm-standby.js";
import type { CloudWarmStandby } from "./services/cloud-warm-standby.js";
import { browserUseRoutes } from "./routes/browser-use.js";
import { registerChatDeliveryWork } from "./services/chat-delivery-work.js";
import { registerBrowserUseCleanup } from "./services/browser-use-work.js";
import { browserUseService } from "./services/browser-use.js";
import { slackToolRoutes } from "./routes/slack-tools.js";
@@ -319,7 +320,7 @@ export function createChatReconciliationCoordinator(input: {
processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
projectRunMilestones: () => Promise<number>;
flushPublications: () => Promise<unknown>;
processPendingSlackFileUploadReceipts: () => Promise<unknown>;
processPendingSlackFileUploadReceipts?: () => Promise<unknown>;
processPendingSlackSessionSyncs: () => Promise<unknown>;
onError: (lane: ChatReconciliationLane, error: unknown) => void;
}) {
@@ -366,7 +367,7 @@ export function createChatReconciliationCoordinator(input: {
}
milestoneReconciliation.poll();
publicationReconciliation.poll();
start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts);
start("Slack session status", input.processPendingSlackSessionSyncs);
},
notifyPublications() {
@@ -1232,25 +1233,24 @@ export async function createApp(
hasPending: () => opts.feedbackExportService!.hasPendingFeedbackTraces(),
});
}
registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
emailChannels.start();
const flushChatPublications = async () => {
await chatChannels.schedulePendingPublications();
const reconcileChatPublicationMaintenance = async () => {
await chatChannels.processPublicationMaintenance();
};
const chatReconciliation = createChatReconciliationCoordinator({
reconcileProviderRuntimes: async () => {
await chatChannels.reconcileProviderRuntimes();
await chatChannels.voice.reconcile();
},
processPendingDeliveries: () => chatChannels.processPendingDeliveries(),
processPendingDeliveries: () => chatChannels.processPendingChatMaintenance(),
processFailedGitHubWebhookDeliveries: () =>
chatChannels.processFailedGitHubWebhookDeliveries(),
projectRunMilestones: () =>
enqueueChatRunMilestones(db, {
publicBaseUrl: opts.authPublicBaseUrl,
}),
flushPublications: () => flushChatPublications(),
processPendingSlackFileUploadReceipts: () =>
chatChannels.processPendingSlackFileUploadReceipts(),
flushPublications: () => reconcileChatPublicationMaintenance(),
processPendingSlackSessionSyncs: () =>
chatChannels.processPendingSlackSessionSyncs(),
onError: (lane, err) => {
@@ -1263,6 +1263,8 @@ export async function createApp(
chatReconciliation.notifyPublications();
},
);
// Provider/action maintenance and milestone projection still use this cadence.
// Inbox, outbox, and file receipt queues have independent commit/deadline wakes.
let chatPublicationTimer: ReturnType<typeof setInterval> | null = setInterval(
() => {
if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile();
server/src/services/chat-channels.ts
+1058 -740
View File
File diff suppressed because it is too large. Load diff
server/src/services/chat-delivery-work.ts
+44
View File
@@ -0,0 +1,44 @@
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
type ChatDeliveryService = {
processQueuedDeliveries(): Promise<unknown>;
scheduleQueuedPublications(): Promise<unknown>;
processPendingSlackFileUploadReceipts(): Promise<unknown>;
nextInboundDeliveryAt(): Promise<number | null>;
nextPublicationAt(): Promise<number | null>;
nextSlackReceiptAt(): Promise<number | null>;
onPublicationsSettled(wake: () => void): void;
};
/** Independent queue workers share the app's scheduler and DB transaction owner. */
export function registerChatDeliveryWork(
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
service: ChatDeliveryService,
canRun: () => boolean,
) {
const tasks = [
{ queue: DELIVERY_QUEUES.chatInbound, run: service.processQueuedDeliveries, next: service.nextInboundDeliveryAt },
{ queue: DELIVERY_QUEUES.chatPublications, run: service.scheduleQueuedPublications, next: service.nextPublicationAt },
{ queue: DELIVERY_QUEUES.chatReceipts, run: service.processPendingSlackFileUploadReceipts, next: service.nextSlackReceiptAt },
];
const workers = tasks.map(task => {
let next: number | null = null;
const worker = coordinator.register(task.queue, {
retryMs: 1000,
run: () => task.run(),
hasPending: async () => {
next = canRun() ? await task.next() : Date.now() + 1000;
return false;
},
nextRunAt: () => next === null ? null : Math.max(Date.now() + 1000, next),
});
if (task.queue === DELIVERY_QUEUES.chatPublications) {
// Dispatch returns after reserving endpoint slots. Refill them when a
// provider task ends, including failures before its final durable write.
service.onPublicationsSettled(worker.wake);
}
return worker;
});
return { ready: Promise.all(workers.map(worker => worker.ready)) };
}
server/src/services/chat-interaction-publications.ts
+148 -133
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
import { nativePhotonInteraction } from "./photon/interactions.js";
import { randomBytes } from "node:crypto";
import { and, eq, inArray, sql } from "drizzle-orm";
@@ -30,7 +31,7 @@ const QUESTION_ACTION_TOKEN_BYTES = 16;
export const CHAT_QUESTION_ACTION_TOKEN_TTL_MS = 7 * 24 * 60 * 60 * 1_000;
export const TELEGRAM_CALLBACK_DATA_LIMIT_BYTES = 64;
type ChatPublicationDb = Pick<Db, "select" | "insert" | "update">;
type ChatPublicationDb = Pick<Db, "select" | "insert" | "update" | "transaction">;
function terminalNativeInteractionCopy(
interaction: IssueThreadInteraction,
@@ -408,83 +409,87 @@ export async function enqueueIssueInteractionChatPublications(
},
},
});
const rows = await db
.insert(chatPublications)
.values({
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
issueId: interaction.issueId,
idempotencyKey: `interaction:${interaction.id}:${endpoint.id}`,
payload,
state: "pending",
})
.onConflictDoNothing()
.returning();
const publication = rows[0];
if (publication && endpoint.provider === "imessage-photon" && nativePhotonInteraction(interaction)) {
const reference = randomBytes(9).toString("base64url");
await db.insert(chatActions).values({ companyId: interaction.companyId, endpointId: endpoint.id, conversationId: conversation.id,
kind: "photon_interaction", providerActionId: `photon:${reference}`,
payload: { version: 1, reference, interactionId: interaction.id, publicationId: publication.id, sessionGeneration: conversation.sessionGeneration,
expiresAt: new Date(publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS).toISOString() }, status: "issued" });
} else if (publication && formDraft) {
await db.insert(chatActions).values(
chatQuestionFormActionRecords(formDraft, {
const rows = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
const rows = await tx
.insert(chatPublications)
.values({
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
publicationId: publication.id,
}),
);
} else if (publication && question && questionActionTokens.length > 0) {
const expiresAt = new Date(
publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
).toISOString();
await db.insert(chatActions).values(
questionActionTokens.map(({ actionId, option }) => ({
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
kind: "question_answer",
providerActionId: actionId,
payload: {
version: 1,
issueId: interaction.issueId,
idempotencyKey: `interaction:${interaction.id}:${endpoint.id}`,
payload,
state: "pending",
})
.onConflictDoNothing()
.returning();
const publication = rows[0];
if (publication && endpoint.provider === "imessage-photon" && nativePhotonInteraction(interaction)) {
const reference = randomBytes(9).toString("base64url");
await tx.insert(chatActions).values({ companyId: interaction.companyId, endpointId: endpoint.id, conversationId: conversation.id,
kind: "photon_interaction", providerActionId: `photon:${reference}`,
payload: { version: 1, reference, interactionId: interaction.id, publicationId: publication.id, sessionGeneration: conversation.sessionGeneration,
expiresAt: new Date(publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS).toISOString() }, status: "issued" });
} else if (publication && formDraft) {
await tx.insert(chatActions).values(
chatQuestionFormActionRecords(formDraft, {
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
publicationId: publication.id,
interactionId: interaction.id,
questionId: question.id,
optionId: option.id,
expiresAt,
},
status: "issued",
})),
);
} else if (
publication &&
confirmation &&
confirmationActionTokens.length > 0
) {
const expiresAt = new Date(
publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
).toISOString();
await db.insert(chatActions).values(
confirmationActionTokens.map(({ actionId, decision }) => ({
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
kind: "confirmation_response",
providerActionId: actionId,
payload: {
version: 1,
publicationId: publication.id,
interactionId: interaction.id,
decision,
expiresAt,
},
status: "issued",
})),
);
}
}),
);
} else if (publication && question && questionActionTokens.length > 0) {
const expiresAt = new Date(
publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
).toISOString();
await tx.insert(chatActions).values(
questionActionTokens.map(({ actionId, option }) => ({
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
kind: "question_answer",
providerActionId: actionId,
payload: {
version: 1,
publicationId: publication.id,
interactionId: interaction.id,
questionId: question.id,
optionId: option.id,
expiresAt,
},
status: "issued",
})),
);
} else if (
publication &&
confirmation &&
confirmationActionTokens.length > 0
) {
const expiresAt = new Date(
publication.createdAt.getTime() + CHAT_QUESTION_ACTION_TOKEN_TTL_MS,
).toISOString();
await tx.insert(chatActions).values(
confirmationActionTokens.map(({ actionId, decision }) => ({
companyId: interaction.companyId,
endpointId: endpoint.id,
conversationId: conversation.id,
kind: "confirmation_response",
providerActionId: actionId,
payload: {
version: 1,
publicationId: publication.id,
interactionId: interaction.id,
decision,
expiresAt,
},
status: "issued",
})),
);
}
return rows;
});
inserted.push(...rows);
}
return inserted;
@@ -560,20 +565,23 @@ export async function enqueueTerminalIssueInteractionChatPublications(
)
.map((publication) => publication.id);
if (unsentIds.length > 0) {
await db
.update(chatPublications)
.set({
state: "cancelled",
nextAttemptAt: null,
redactedError: "Interaction was resolved before provider publication",
updatedAt: new Date(),
})
.where(
and(
inArray(chatPublications.id, unsentIds),
inArray(chatPublications.state, ["pending", "retry"]),
),
);
await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatPublications)
.set({
state: "cancelled",
nextAttemptAt: null,
redactedError: "Interaction was resolved before provider publication",
updatedAt: new Date(),
})
.where(
and(
inArray(chatPublications.id, unsentIds),
inArray(chatPublications.state, ["pending", "retry"]),
),
);
});
}
const originalIds = originals.map((publication) => publication.id);
@@ -706,33 +714,37 @@ export async function enqueueTerminalIssueInteractionChatPublications(
if (!copy) return [];
const inserted: Array<typeof chatPublications.$inferSelect> = [];
for (const original of providerVisibleOriginals) {
if (!original.payload.card) continue;
const rows = await db
.insert(chatPublications)
.values({
companyId: interaction.companyId,
endpointId: original.endpointId,
conversationId: original.conversationId,
issueId: interaction.issueId,
idempotencyKey: `interaction-resolution:${interaction.id}:${original.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "issue_interaction",
text: copy.text,
interaction: {
id: interaction.id,
card: {
kind: original.payload.card.kind,
title: original.payload.card.title,
body: copy.body,
actions: [],
const card = original.payload.card;
if (!card) continue;
const rows = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.insert(chatPublications)
.values({
companyId: interaction.companyId,
endpointId: original.endpointId,
conversationId: original.conversationId,
issueId: interaction.issueId,
idempotencyKey: `interaction-resolution:${interaction.id}:${original.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "issue_interaction",
text: copy.text,
interaction: {
id: interaction.id,
card: {
kind: card.kind,
title: card.title,
body: copy.body,
actions: [],
},
},
},
}),
state: "pending",
})
.onConflictDoNothing()
.returning();
}),
state: "pending",
})
.onConflictDoNothing()
.returning();
});
inserted.push(...rows);
}
return inserted;
@@ -771,23 +783,26 @@ export async function cancelPendingIssueInteractionChatPublications(
]),
),
);
return db
.update(chatPublications)
.set({
state: "cancelled",
nextAttemptAt: null,
redactedError: "Interaction was superseded before publication",
updatedAt: new Date(),
})
.where(
and(
eq(chatPublications.companyId, input.companyId),
eq(chatPublications.issueId, input.issueId),
inArray(chatPublications.state, ["pending", "retry"]),
inArray(sql<string>`${chatPublications.payload}->>'interactionId'`, [
...input.interactionIds,
]),
),
)
.returning();
return db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatPublications)
.set({
state: "cancelled",
nextAttemptAt: null,
redactedError: "Interaction was superseded before publication",
updatedAt: new Date(),
})
.where(
and(
eq(chatPublications.companyId, input.companyId),
eq(chatPublications.issueId, input.issueId),
inArray(chatPublications.state, ["pending", "retry"]),
inArray(sql<string>`${chatPublications.payload}->>'interactionId'`, [
...input.interactionIds,
]),
),
)
.returning();
});
}
server/src/services/chat-run-publications.ts
+28 -23
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
import {
and,
asc,
@@ -623,6 +624,7 @@ async function enqueueSafeNativeChatProgress(
.limit(1);
if (explicitlyAuthoredFinal.length > 0) return 0;
await notifyChatPublicationWork(tx);
const insertedRows = await tx
.insert(chatPublications)
.values({
@@ -997,30 +999,33 @@ export async function enqueueChatRunMilestones(
.limit(1);
if (explicitlyAuthoredPublication.length > 0) continue;
}
const result = await db
.insert(chatPublications)
.values({
companyId: row.companyId,
endpointId: row.endpointId,
conversationId: row.conversationId,
issueId: row.issueId,
idempotencyKey: `run:${row.runId}:${milestone}:${row.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "safe_milestone",
text: safeMilestoneText({
agentName: row.agentName,
errorCode: row.runErrorCode,
milestone,
issueId: row.issueId,
publicBaseUrl: input.publicBaseUrl,
const result = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.insert(chatPublications)
.values({
companyId: row.companyId,
endpointId: row.endpointId,
conversationId: row.conversationId,
issueId: row.issueId,
idempotencyKey: `run:${row.runId}:${milestone}:${row.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "safe_milestone",
text: safeMilestoneText({
agentName: row.agentName,
errorCode: row.runErrorCode,
milestone,
issueId: row.issueId,
publicBaseUrl: input.publicBaseUrl,
}),
progressState: milestone,
}),
progressState: milestone,
}),
state: "pending",
})
.onConflictDoNothing()
.returning({ id: chatPublications.id });
state: "pending",
})
.onConflictDoNothing()
.returning({ id: chatPublications.id });
});
inserted += result.length;
}
if (rows.length < pageSize) break;
server/src/services/chat-teams-file-publication.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
import { and, eq, sql } from "drizzle-orm";
import { z } from "zod";
import { CHAT_FILE_TRANSFER_PHASES } from "@paperclipai/shared";
@@ -375,6 +376,7 @@ export async function projectTeamsFilePublication(
!sameDate(publication.nextAttemptAt, nextAttemptAt) ||
!sameDate(publication.publishedAt, publishedAt)
) {
await notifyChatPublicationWork(tx);
await tx
.update(chatPublications)
.set({
server/src/services/chat-work-notifications.ts
+16
View File
@@ -0,0 +1,16 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
export async function notifyChatDeliveryWork(tx: object): Promise<void> {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInbound);
}
export async function notifyChatPublicationWork(tx: object): Promise<void> {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatPublications);
// Receipt recovery waits for the publication owner to release its claim.
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceipts);
}
export async function notifyChatEndpointWork(tx: object): Promise<void> {
await notifyChatDeliveryWork(tx);
await notifyChatPublicationWork(tx);
}
server/src/services/connectors/slack-delivery.ts
+28 -21
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "../chat-work-notifications.js";
import { and, eq, sql, inArray } from "drizzle-orm";
import { chatActions, type Db } from "@paperclipai/db";
import { forbidden } from "../../errors.js";
@@ -40,17 +41,20 @@ export async function slackDelivery(
action.status === "processing" &&
Date.now() - action.updatedAt.getTime() > 5 * 60_000
) {
const [stale] = await db
.update(chatActions)
.set({ status: "uncertain", updatedAt: new Date() })
.where(
and(
eq(chatActions.id, action.id),
eq(chatActions.status, "processing"),
eq(chatActions.updatedAt, action.updatedAt),
),
)
.returning();
const [stale] = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatActions)
.set({ status: "uncertain", updatedAt: new Date() })
.where(
and(
eq(chatActions.id, action.id),
eq(chatActions.status, "processing"),
eq(chatActions.updatedAt, action.updatedAt),
),
)
.returning();
});
if (stale) action = stale;
}
const args = object(action.payload.args);
@@ -107,16 +111,19 @@ export async function slackDelivery(
};
}
if (receipt) {
const [settled] = await db
.update(chatActions)
.set({ status: "processed", result: receipt, updatedAt: new Date() })
.where(
and(
eq(chatActions.id, action.id),
inArray(chatActions.status, ["uncertain", "processing"]),
),
)
.returning();
const [settled] = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatActions)
.set({ status: "processed", result: receipt, updatedAt: new Date() })
.where(
and(
eq(chatActions.id, action.id),
inArray(chatActions.status, ["uncertain", "processing"]),
),
)
.returning();
});
if (settled) action = settled;
}
}
server/src/services/connectors/slack-publication.ts
+48 -75
View File
@@ -1,81 +1,54 @@
import { and, eq, sql, inArray } from "drizzle-orm";
import {
chatActions,
chatConversations,
issueComments,
type chatPublications,
type Db,
} from "@paperclipai/db";
import { object } from "./slack-client.js";
import { sql, type SQL } from "drizzle-orm";
import { chatPublications, type Db } from "@paperclipai/db";
/** Explicit sends use a separate durable receipt. Suppress an identical automatic
* final, but retain distinct summaries, questions, blockers and progress. */
// ECMAScript String.trim whitespace, matching tool-send text comparison.
const trimCharacters = "\u0009\u000a\u000b\u000c\u000d\u0020\u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000\ufeff";
type PublicationMatch = {
companyId: SQL; endpointId: SQL; conversationId: SQL;
issueId: SQL; commentId: SQL; payload: SQL;
};
function matchingSlackSend(publication: PublicationMatch) {
const p = publication;
return sql`from chat_actions send
join issue_comments comment on comment.id = ${p.commentId}
and comment.company_id = ${p.companyId} and comment.issue_id = ${p.issueId}
join chat_conversations conversation on conversation.id = ${p.conversationId}
and conversation.company_id = ${p.companyId}
where send.company_id = ${p.companyId} and send.endpoint_id = ${p.endpointId}
and send.conversation_id = ${p.conversationId} and send.kind = 'slack_tool_write'
and send.payload->'binding'->>'runId' = coalesce(comment.created_by_run_id, comment.derived_created_by_run_id)::text
and send.payload->>'name' = 'slack_post_message'
and coalesce(${p.payload}->>'interactionId', '') = ''
and coalesce(${p.payload}->>'progressState', '') = ''
and (case when jsonb_typeof(${p.payload}->'attachmentIds') = 'array'
then jsonb_array_length(${p.payload}->'attachmentIds') else 0 end) = 0
and send.payload->'args'->>'channel' = regexp_replace(conversation.external_conversation_id, '^slack:', '')
and send.payload->'args'->>'thread_ts' = regexp_replace(conversation.external_thread_id, '^.*:', '')
and btrim(send.payload->'args'->>'text', ${trimCharacters}) = btrim(${p.payload}->>'text', ${trimCharacters})`;
}
/** Dispatch and deadlines must agree about pending explicit Slack sends. */
export function unresolvedSlackPublicationCondition(): SQL {
return sql`exists (select 1 ${matchingSlackSend({
companyId: sql`${chatPublications.companyId}`, endpointId: sql`${chatPublications.endpointId}`,
conversationId: sql`${chatPublications.conversationId}`, issueId: sql`${chatPublications.issueId}`,
commentId: sql`${chatPublications.commentId}`, payload: sql`${chatPublications.payload}`,
})} and send.status in ('received', 'processing', 'uncertain'))`;
}
/** Explicit sends suppress identical automatic finals, but not distinct content. */
export async function slackExplicitPublicationDuplicate(
db: Db,
publication: typeof chatPublications.$inferSelect,
) {
if (
!publication.commentId ||
publication.payload.interactionId ||
publication.payload.progressState ||
publication.payload.attachmentIds?.length
)
return null;
const [comment] = await db
.select({
runId: sql<string>`coalesce(${issueComments.createdByRunId}, ${issueComments.derivedCreatedByRunId})`,
})
.from(issueComments)
.where(
and(
eq(issueComments.id, publication.commentId),
eq(issueComments.companyId, publication.companyId),
eq(issueComments.issueId, publication.issueId),
),
);
if (!comment?.runId) return null;
const [conversation] = await db
.select()
.from(chatConversations)
.where(
and(
eq(chatConversations.id, publication.conversationId),
eq(chatConversations.companyId, publication.companyId),
),
);
if (!conversation) return null;
const sends = await db
.select({ payload: chatActions.payload, status: chatActions.status })
.from(chatActions)
.where(
and(
eq(chatActions.companyId, publication.companyId),
eq(chatActions.endpointId, publication.endpointId),
eq(chatActions.conversationId, publication.conversationId),
eq(chatActions.kind, "slack_tool_write"),
inArray(chatActions.status, [
"processed",
"received",
"processing",
"uncertain",
]),
sql`${chatActions.payload}->'binding'->>'runId' = ${comment.runId}`,
sql`${chatActions.payload}->>'name' = 'slack_post_message'`,
),
);
const match = sends.find(({ payload }) => {
const args = object(payload.args);
const thread = conversation.externalThreadId.split(":").at(-1);
return (
args.channel ===
conversation.externalConversationId.replace(/^slack:/, "") &&
args.thread_ts === thread &&
String(args.text).trim() === publication.payload.text.trim()
);
});
return match
? match.status === "processed"
? "delivered"
: "unresolved"
: null;
const p = publication;
if (!p.commentId || p.payload.interactionId || p.payload.progressState || p.payload.attachmentIds?.length) return null;
const rows = await db.execute<{ status: string }>(sql`select send.status ${matchingSlackSend({
companyId: sql`${p.companyId}::uuid`, endpointId: sql`${p.endpointId}::uuid`,
conversationId: sql`${p.conversationId}::uuid`, issueId: sql`${p.issueId}::uuid`,
commentId: sql`${p.commentId}::uuid`, payload: sql`${JSON.stringify(p.payload)}::jsonb`,
})} and send.status in ('processed', 'received', 'processing', 'uncertain')
order by (send.status = 'processed') asc limit 1`);
return rows[0] ? rows[0].status === "processed" ? "delivered" : "unresolved" : null;
}
server/src/services/connectors/slack-retry.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "../chat-work-notifications.js";
import { and, eq, sql, inArray } from "drizzle-orm";
import { chatActions, toolInvocations, type Db } from "@paperclipai/db";
import {
@@ -52,6 +53,7 @@ export async function claimSlackRateLimitRetry(
!(Date.parse(action.result.retryAt) <= Date.now())
)
return false;
await notifyChatPublicationWork(tx);
await tx
.update(chatActions)
.set({ status: "received", updatedAt: new Date() })
server/src/services/connectors/slack-writes.ts
+69 -53
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "../chat-work-notifications.js";
import { authorizeSlackDocument } from "./slack-document.js";
import { slackMessage } from "./slack-message.js";
import { createHash } from "node:crypto";
@@ -100,28 +101,31 @@ export async function executeSlackWrite(
.update(JSON.stringify({ name, args }))
.digest("hex");
const key = `slack-tool:${binding.issueId}:${args.idempotencyKey}`;
const [created] = await db
.insert(chatActions)
.values({
companyId: binding.companyId,
endpointId: authority.endpoint.id,
conversationId: authority.conversation?.id ?? null,
principalId: authority.principalId,
kind: "slack_tool_write",
providerActionId: key,
status: "received",
payload: {
name,
args,
hash,
binding,
userId: authority.userId,
revision: authority.revision,
invocationId,
},
})
.onConflictDoNothing()
.returning();
const [created] = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.insert(chatActions)
.values({
companyId: binding.companyId,
endpointId: authority.endpoint.id,
conversationId: authority.conversation?.id ?? null,
principalId: authority.principalId,
kind: "slack_tool_write",
providerActionId: key,
status: "received",
payload: {
name,
args,
hash,
binding,
userId: authority.userId,
revision: authority.revision,
invocationId,
},
})
.onConflictDoNothing()
.returning();
});
const action =
created ??
(
@@ -151,13 +155,16 @@ export async function executeSlackWrite(
instruction:
"Do not retry with another key. Inspect delivery before reconciling an uncertain result.",
};
const [claimed] = await db
.update(chatActions)
.set({ status: "processing", updatedAt: new Date() })
.where(
and(eq(chatActions.id, action.id), eq(chatActions.status, "received")),
)
.returning();
const [claimed] = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatActions)
.set({ status: "processing", updatedAt: new Date() })
.where(
and(eq(chatActions.id, action.id), eq(chatActions.status, "received")),
)
.returning();
});
if (!claimed) return { actionId: action.id, state: "processing" };
try {
authority = await resolveSlackTaskAuthority(db, binding);
@@ -252,13 +259,16 @@ export async function executeSlackWrite(
)
throw forbidden("Slack returned an unsupported upload destination");
// Store the provider receipt before transport. An uncertain finish is never blindly retried.
await db
.update(chatActions)
.set({
result: { fileId: upload.file_id, phase: "upload_allocated" },
updatedAt: new Date(),
})
.where(eq(chatActions.id, action.id));
await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatActions)
.set({
result: { fileId: upload.file_id, phase: "upload_allocated" },
updatedAt: new Date(),
})
.where(eq(chatActions.id, action.id));
});
const uploaded = await fetchImpl(url, {
method: "POST",
body: bytes,
@@ -436,10 +446,13 @@ export async function executeSlackWrite(
}
: {}),
};
await db
.update(chatActions)
.set({ status: "processed", result: receipt, updatedAt: new Date() })
.where(eq(chatActions.id, action.id));
await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatActions)
.set({ status: "processed", result: receipt, updatedAt: new Date() })
.where(eq(chatActions.id, action.id));
});
await logActivity(db, {
companyId: binding.companyId,
actorType: "agent",
@@ -468,19 +481,22 @@ export async function executeSlackWrite(
"slack_request_timeout",
"slack_service_unavailable",
].includes(details.code)));
await db
.update(chatActions)
.set({
status: definite ? "failed" : "uncertain",
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify({ code: details.code ?? "slack_delivery_uncertain", ...(error instanceof HttpError && error.status === 429 ? { retryAt: new Date(Date.now() + Number(details.retryAfterSeconds ?? 60) * 1000).toISOString() } : {}) })}::jsonb`,
updatedAt: new Date(),
})
.where(
and(
eq(chatActions.id, action.id),
eq(chatActions.status, "processing"),
),
);
await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
return tx
.update(chatActions)
.set({
status: definite ? "failed" : "uncertain",
result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || ${JSON.stringify({ code: details.code ?? "slack_delivery_uncertain", ...(error instanceof HttpError && error.status === 429 ? { retryAt: new Date(Date.now() + Number(details.retryAfterSeconds ?? 60) * 1000).toISOString() } : {}) })}::jsonb`,
updatedAt: new Date(),
})
.where(
and(
eq(chatActions.id, action.id),
eq(chatActions.status, "processing"),
),
);
});
if (error instanceof HttpError)
error.details = {
...details,
server/src/services/delivery-work-notifications.ts
+3
View File
@@ -2,6 +2,9 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
/** Topics for existing durable queues; register intent before their writes. */
export const DELIVERY_QUEUES = {
chatInbound: "chat-inbound",
chatPublications: "chat-publications",
chatReceipts: "chat-slack-receipts",
browser: "browser-use",
email: "email-channels",
feedback: "feedback-exports",
server/src/services/issues.ts
+38 -30
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
import { recordChatHandoff, recordChatCompletion, existingChatCompletionReply, acknowledgeChatCompletionReply } from "./chat-completion-delivery.js";
import { mirrorSlackBoardComment, slackBoardReplyBindings } from "./slack-board-messages.js";
import { assertAgentRunWriteAllowed } from "../agent-run-cancellation.js";
@@ -13047,49 +13048,55 @@ export function issueService(db: Db) {
);
const publicationCreatedAt = new Date();
for (const binding of bindings) {
await dbOrTx
.insert(chatPublications)
.values({
companyId: binding.companyId,
endpointId: binding.endpointId,
conversationId: binding.conversationId,
issueId,
commentId: comment.id,
idempotencyKey: `comment:${comment.id}:${binding.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "agent_comment",
text: redactedBody,
}),
state: "pending",
createdAt: publicationCreatedAt,
updatedAt: publicationCreatedAt,
})
.onConflictDoNothing();
for (const [index, attachment] of boundAttachments.entries()) {
const attachmentCreatedAt = new Date(
publicationCreatedAt.getTime() + index + 1,
);
await dbOrTx
await dbOrTx.transaction(async (tx: DbTransaction) => {
await notifyChatPublicationWork(tx);
return tx
.insert(chatPublications)
.values({
companyId: binding.companyId,
endpointId: binding.endpointId,
conversationId: binding.conversationId,
issueId,
commentId: attachment.commentId,
idempotencyKey: `attachment:${attachment.id}:${binding.endpointId}`,
commentId: comment.id,
idempotencyKey: `comment:${comment.id}:${binding.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "agent_comment",
text: `Shared ${attachment.originalFilename ?? "a file"}.`,
attachmentIds: [attachment.id],
text: redactedBody,
}),
state: "pending",
createdAt: attachmentCreatedAt,
updatedAt: attachmentCreatedAt,
createdAt: publicationCreatedAt,
updatedAt: publicationCreatedAt,
})
.onConflictDoNothing();
});
for (const [index, attachment] of boundAttachments.entries()) {
const attachmentCreatedAt = new Date(
publicationCreatedAt.getTime() + index + 1,
);
await dbOrTx.transaction(async (tx: DbTransaction) => {
await notifyChatPublicationWork(tx);
return tx
.insert(chatPublications)
.values({
companyId: binding.companyId,
endpointId: binding.endpointId,
conversationId: binding.conversationId,
issueId,
commentId: attachment.commentId,
idempotencyKey: `attachment:${attachment.id}:${binding.endpointId}`,
payload: projectSafeChatPublication({
classification: "external",
source: "agent_comment",
text: `Shared ${attachment.originalFilename ?? "a file"}.`,
attachmentIds: [attachment.id],
}),
state: "pending",
createdAt: attachmentCreatedAt,
updatedAt: attachmentCreatedAt,
})
.onConflictDoNothing();
});
}
}
}
@@ -13333,6 +13340,7 @@ export function issueService(db: Db) {
registeredRunId,
);
for (const binding of bindings) {
await notifyChatPublicationWork(tx);
await tx
.insert(chatPublications)
.values({
server/src/services/slack-board-messages.ts
+4 -1
View File
@@ -1,3 +1,4 @@
import { notifyChatPublicationWork } from "./chat-work-notifications.js";
import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
import {
assets,
@@ -121,6 +122,7 @@ export async function mirrorSlackBoardComment(
.onConflictDoNothing()
.returning();
if (!receipt) continue;
await notifyChatPublicationWork(db);
const [publication] = await db
.insert(chatPublications)
.values({
@@ -161,7 +163,7 @@ export async function mirrorSlackBoardComment(
byId.has(id) ? [byId.get(id)!] : [],
)
: files;
for (const [index, file] of orderedFiles.entries())
for (const [index, file] of orderedFiles.entries()) {
await db
.insert(chatPublications)
.values({
@@ -181,6 +183,7 @@ export async function mirrorSlackBoardComment(
createdAt: new Date(Date.now() + index + 1),
})
.onConflictDoNothing();
}
await logActivity(db as Db, {
companyId: comment.companyId,
actorType: "user",
server/src/services/voice/voice-session-store.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyChatDeliveryWork } from "../chat-work-notifications.js";
import { environmentService } from "../environments.js";
import { logger } from "../../middleware/logger.js";
import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
@@ -210,6 +211,7 @@ export function voiceSessionStore(db: Db, options: { allowLocalBoard: boolean; n
const threadId = conversation.externalThreadId;
const providerEventId = `${threadId}:${session.id}:${toolId}`;
const trigger = prior.length ? "subscribed_message" : "direct_message";
await notifyChatDeliveryWork(tx);
const [delivery] = await tx.insert(chatDeliveries).values({
companyId: session.companyId, endpointId: session.endpointId, conversationId: session.conversationId,
providerEventId, deduplicationKey: createHash("sha256").update(providerEventId).digest("hex"),