diff --git a/doc/performance/2026-11-03-response-presentation-contention.md b/doc/performance/2026-11-03-response-presentation-contention.md new file mode 100644 index 00000000000..735452d7c0c --- /dev/null +++ b/doc/performance/2026-11-03-response-presentation-contention.md @@ -0,0 +1,74 @@ +# Avoid self-contention during accepted-response recovery + +Baseline: `82c033ed127d07862e6765aa4b69a3cf339bde81`. This change prevents the response-presentation worker from recovering different tasks on a shared chat endpoint or principal at the same time. It preserves task order, current transactional authorization, company boundaries, NOWAIT locks, and retry/drain behavior. Endpoint and principal IDs are scheduling hints from a company-scoped wake-action join. Missing hints retain task order and every eligibility check. + +## Evidence and limits + +A staging log from the captured host showed `native-response-presentation could not settle` → DrizzleQueryError → PostgresError `could not obtain lock on row in relation "chat_endpoints"`. The retained hosted sample did not include SQLSTATE. Repeated presentation errors establish activity, not its CPU share. Most SQL-construction samples in the fresh profile lack application ancestors. This change makes no claim to resolve all main-loop CPU. + +The real PostgreSQL regression uses four accepted historical responses for distinct tasks and principals on one actual chat endpoint. It uses the live source-authority implementation. The baseline worker produced three endpoint NOWAIT failures in one pass without external ingress. The new admission settles all four responses exactly once. An owned advisory gate holds a real comment transaction in the correctness test. An independent endpoint must publish before that gate is released. + +Other controls cover external-lock failure and rollback, recovery after release, paused-endpoint refusal despite a forged hint, transitive task/endpoint groups, duplicate and missing hints, shutdown, drain/resume, and an in-flight rescope with exactly-once newcomer admission. All authorization locks remain in their owners. Ingress, direct finalization/reconciliation, and other workers can still cause legitimate retries. Provider network sends are not asserted to hold these locks. Open PR #17714 fixes a complementary publication-commit lock order. PR #18686 changes run-list JSON/index work, not this worker. + +## Initial endpoint-only comparison + +Five alternating pairs compared the actual archived baseline worker/query with the initial endpoint-only treatment on source base `82c033ed`. This study preceded the principal-hint review fix below. It measures the earlier candidate, not final-head CPU. The final endpoint fixture has four distinct principals, so the added principal key does not change its groups; its constant extra hint processing was not CPU-measured. Each used fresh equivalent four-response fixtures and real PostgreSQL. No artificial database sleep or advisory gate was used in this comparison. Validation checked one comment per run, the exact accepted text, and one publication with the matching text/company/task/conversation/comment association. Validation ran after timing and capture counters stopped. + +The actual Vitest child recorded Node **v24.22.2** and executable `/Users/dotta/.npm/_npx/387698761821791d/node_modules/node/bin/node`. This is a shared-host measurement: the coordinator reported 18 logical CPUs and about 10.5 one-minute load, with unrelated Node/VM work. Other campaign agents paused heavy jobs. No unrelated process was stopped. + +CPU/wall measurement covers worker start through settlement. It excludes imports, fixture creation, and output validation. Existing capture hooks count product driver calls, returned rows, and errors before Drizzle decoding. Completion observation reads only in-memory idle state; it issues **no SQL**. CPU includes capture hooks, timer processing, and the unequal number of observer checks shown below. It is instrumented Vitest-process CPU, not pure product CPU or PostgreSQL CPU. The outlier is retained. + +Retry timers and Date were advanced together. **Measured wall time excludes real retry waits** and is not a hosted latency claim. Baseline virtual elapsed time was 92–211 seconds, with two or three retry deadlines replayed. Treatment replayed none. Raw driver calls include transaction-control statements. These results demonstrate avoided local retry work; no hosted CPU percentage reduction is claimed. + +| Round | Variant | CPU ms | Measured wall ms | Driver calls | Rows | Query errors | Retry deadlines | Memory checks | Virtual elapsed ms | +| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | +| 1 | before | 169.487 | 4263.348 | 393 | 276 | 4 | 2 | 202 | 92082 | +| 1 | after | 80.695 | 1058.891 | 293 | 200 | 0 | 0 | 45 | 1020 | +| 2 | after | 121.239 | 1066.592 | 293 | 200 | 0 | 0 | 45 | 1020 | +| 2 | before | 141.654 | 4261.701 | 443 | 314 | 6 | 3 | 205 | 211063 | +| 3 | before | 945.351 | 4321.276 | 443 | 314 | 6 | 3 | 205 | 211083 | +| 3 | after | 100.563 | 1075.898 | 293 | 200 | 0 | 0 | 44 | 1020 | +| 4 | after | 83.766 | 1076.359 | 293 | 200 | 0 | 0 | 44 | 1020 | +| 4 | before | 534.683 | 4304.004 | 443 | 314 | 6 | 3 | 205 | 211083 | +| 5 | before | 177.157 | 4377.283 | 393 | 276 | 4 | 2 | 201 | 92102 | +| 5 | after | 98.164 | 1072.052 | 293 | 200 | 0 | 0 | 40 | 1020 | + +Median instrumented process CPU was 177.157 → 98.164 ms, with ranges 141.654–945.351 and 80.695–121.239 ms. Median product driver calls were 443 → 293; rows were 314 → 200; query errors were 6 → 0. Each failed endpoint attempt added exactly 25 product calls and 19 returned rows. All variants completed the same four responses. The observer performed 201–205 memory checks before and 40–45 after; its CPU contribution is included and is not subtracted. + +## Final shared-principal work-count comparison + +Review identified a second shared lock: the source-authority path takes a company/principal try-advisory lock before endpoint checks. Two valid Slack bot endpoints in the same workspace can share one external-principal row. The final code also groups that principal hint from the same company-scoped wake-action join. The hint never grants authority. + +A real PostgreSQL regression asserts distinct endpoints and identical source principal IDs. Before the fix, one response completed on the first pass and the other failed with `NativeChatReviewPresentationContentionError`. The advisory SQL itself returned successfully, so query-error count was zero. After the fix, both exact responses and their correctly associated publications completed on the first pass. This counts-only experiment used a short real comment-transaction hold to reproduce overlap. It measured no CPU or wall-time benefit. + +| Stage | Admission | Responses | Product calls | Rows | Outer contention failures | Query errors | +| --- | --- | ---: | ---: | ---: | ---: | ---: | +| First pass | Task + endpoint | 1 | 91 | 65 | 1 | 0 | +| First pass | Task + endpoint + principal | 2 | 145 | 100 | 0 | 0 | +| Same completed work, after retry | Task + endpoint | 2 | 165 | 114 | 1 | 0 | +| Same completed work, settled | Task + endpoint + principal | 2 | 147 | 100 | 0 | 0 | + +The final independent-resource control retains parallel progress when task, endpoint and principal keys differ. Existing endpoint-lock and current-authority controls remain. The original five-pair table is retained as historical evidence; no final-code CPU percentage is inferred from it. + +## Reproduction and failure accounting + +```sh +pnpm exec vitest run --project @paperclipai/server server/src/__tests__/scoped-deadline-work.test.ts +pnpm exec vitest run --project @paperclipai/server server/src/__tests__/chat-channels.integration.test.ts -t 'shared-endpoint accepted-response|different endpoints for one principal|endpoint scheduling hint|contended accepted response' +``` + +The committed helper materializes the pinned baseline with `git show`; it does not ship a clone of old production code. Run it with Node 24.22.2 and pnpm 9.15.4 from a clean, isolated checkout of this PR: + +```sh +git worktree add --detach /tmp/presentation-repro HEAD +cd /tmp/presentation-repro +pnpm install --frozen-lockfile +pnpm --filter @paperclipai/plugin-sdk ensure-build-deps +node server/scripts/benchmark-response-presentation.mjs /tmp/presentation-results +``` + +The helper checks the actual child runtime, writes each pair immediately, and restores temporary fixture/baseline files in `finally`. Its output includes the baseline/treatment revisions, exact benchmark fragment, baseline modules, child preflight, JSONL and Vitest log. Review `server/scripts/benchmark-response-presentation.mjs` for the measurement boundary. Initial study artifacts remain under `/private/tmp/paperclip-perf-batch-20261102/campaign-10h/presentation-benchmark/`; the table above makes all ten measured runs (five pairs) inspectable without that workspace. + +Initial offline install lacked a cached tarball; frozen network install passed. The first test import lacked built plugin-sdk; its existing build helper resolved that. The intentionally failing same-endpoint regression is retained. A Date-only benchmark clock failed to settle. A corrected run passed but lost console metrics, so it supplies no numbers. The next retained run used inherited Node 26; it is superseded and is not the table above. The final run checked the actual Node 24 child, verified its file sink, and retained each pair immediately. An early broad check inherited global pnpm 11 and failed its version-switch signature fetch; the corrected checks use the pinned Node 24/pnpm 9 PATH. These setup failures are not reported as verification passes. + +The first full typecheck found an existing direct-worker fixture that omits the endpoint hint. The hint type now explicitly permits omission, and dispatch falls back to the existing task key. A sandbox IPC restriction required rerunning the pinned typecheck with local socket access. diff --git a/server/scripts/benchmark-response-presentation.mjs b/server/scripts/benchmark-response-presentation.mjs new file mode 100644 index 00000000000..21a3fbae89e --- /dev/null +++ b/server/scripts/benchmark-response-presentation.mjs @@ -0,0 +1,133 @@ +// Run only in a clean, isolated checkout. Reuses the real chat integration fixture. +import assert from "node:assert/strict"; +import { existsSync, mkdirSync, mkdtempSync, openSync, closeSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { execFileSync, spawnSync } from "node:child_process"; + +assert.equal(process.version, "v24.22.2", "Use Node24.22.2 for the recorded comparison"); +const root = fileURLToPath(new URL("../..", import.meta.url)); +const output = path.resolve(process.argv[2] ?? mkdtempSync(path.join(tmpdir(), "presentation-benchmark-"))); +mkdirSync(output, { recursive: true }); +assert.equal(existsSync(path.join(output, "pairs.jsonl")), false, "Use a fresh output directory"); +const baseline = "82c033ed127d07862e6765aa4b69a3cf339bde81"; +const integration = "server/src/__tests__/chat-channels.integration.test.ts"; +const git = (...args) => execFileSync("git", args, { cwd: root, encoding: "utf8" }); +assert.equal(git("status", "--porcelain").trim(), "", "Commit fixture changes first; use an isolated checkout"); +const original = readFileSync(path.join(root, integration), "utf8"); +const anchor = ' it("settles a shared-endpoint accepted-response backlog'; +assert.equal(original.split(anchor).length, 2, "Expected one fixture insertion point"); +const files = ["query", "work"].map(kind => `server/src/services/native-runtime/native-response-presentation-${kind}.baseline.ts`); +for (const file of files) assert.equal(existsSync(path.join(root, file)), false, `Refusing to replace ${file}`); +const env = { ...process.env, PATH: `${path.dirname(process.execPath)}${path.delimiter}${process.env.PATH ?? ""}`, + PAPERCLIP_PRESENTATION_BENCHMARK_OUTPUT: output }; +assert.equal(execFileSync("pnpm", ["--version"], { cwd: root, env, encoding: "utf8" }).trim(), "9.15.4", "Use pnpm9.15.4"); +writeFileSync(path.join(output, "metadata.json"), JSON.stringify({ baseline, treatment: git("rev-parse", "HEAD").trim(), + node: process.version, execPath: process.execPath, pairs: 5, runs: 10, + mode: "real PostgreSQL source service under Vitest; accelerated retry timers; instrumented process CPU" }, null, 2)); +const benchmark = String.raw` it("measures equivalent completed accepted-response work without artificial sleeps", async () => { + const { appendFileSync } = await import("node:fs"); + const output = process.env.PAPERCLIP_PRESENTATION_BENCHMARK_OUTPUT; + if (!output) throw new Error("Benchmark output directory is required"); + const { nativeResponsePresentationWork: baselineWork } = await import("../services/native-runtime/native-response-presentation-work.baseline.js"); + const { createDatabaseResponseCapture, installDatabaseResponseCapture } = await import("@paperclipai/db/response-capture"); + const schema = await import("@paperclipai/db"); + appendFileSync(output + "/child-runtime.jsonl", + JSON.stringify({ version: process.version, execPath: process.execPath }) + "\n"); + expect(process.version).toBe("v24.22.2"); + const fixedTime = Date.now(); + const capture = createDatabaseResponseCapture({ start: fixedTime - 1, end: fixedTime + 59_999 }, () => {}, () => fixedTime); + let measuring = false, queries = 0, rows = 0, queryErrors = 0; + capture.execute = (action => { + const counted = measuring; + if (counted) queries++; + return Promise.resolve().then(action).then(value => { + if (counted && Array.isArray(value)) rows += value.length; + return value; + }, error => { if (counted) queryErrors++; throw error; }); + }) as typeof capture.execute; + installDatabaseResponseCapture(db, schema, capture); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + for (let round = 0; round < 5; round++) { + for (const label of round % 2 ? ["after", "before"] : ["before", "after"]) { + const fixture = await seedCompany(); + const configured = await configuredTelegramEndpoint(fixture); + const contexts = []; + for (let index = 0; index < 4; index++) contexts.push(await committedChatResponseRecoveryFixture( + "telegram", "coordinator", false, undefined, undefined, true, + { fixture, configured, suffix: String(311 + index) })); + const failures: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => failures.push(error) }); + queries = rows = queryErrors = 0; + const wallStart = performance.now(), cpuStart = process.cpuUsage(), virtualStart = Date.now(); + measuring = true; + const worker = (label === "before" ? baselineWork : nativeResponsePresentationWork)(db).register(coordinator, () => true); + let passes = 0, observerIterations = 0; + try { + await worker.ready; + for (;;) { + while (idleWorkSnapshot().active !== 0) { + observerIterations++; + await new Promise(resolve => setTimeout(resolve, 1)); + } + observerIterations++; + const next = coordinator.nextWakeAt(); + if (next === null) break; + if (++passes > 10) throw new Error("Response benchmark did not settle"); + await vi.advanceTimersByTimeAsync(Math.max(0, next - Date.now()) + 1); + } + measuring = false; + const cpu = process.cpuUsage(cpuStart), wallMs = performance.now() - wallStart; + for (const context of contexts) { + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId)); + expect(comments).toHaveLength(1); + expect(comments[0].body).toBe(context.result.summary); + const publications = await readChatPublicationsByCommentId(comments[0].id); + expect(publications).toHaveLength(1); + expect(publications[0]).toMatchObject({ companyId: fixture.companyId, endpointId: configured.endpoint.id, + issueId: context.issue.id, conversationId: context.conversation.id, payload: { text: context.result.summary } }); + } + expect(await readChatPublicationsByEndpointId(configured.endpoint.id)).toHaveLength(4); + appendFileSync(output + "/pairs.jsonl", JSON.stringify({ round, label, node: process.version, execPath: process.execPath, observerIterations, + wallMs, cpuMs: (cpu.user + cpu.system) / 1000, queries, rows, queryErrors, + failedPasses: failures.length, replayedRetryDeadlines: passes, + virtualElapsedMs: Date.now() - virtualStart, completedResponses: 4, + artificialDatabaseSleep: false, endpointCount: 1, issueCount: 4 }) + "\n"); + } finally { measuring = false; worker.stop(); await coordinator.stop(); await configured.service.shutdown(); } + } + } + } finally { measuring = false; capture.flush(); vi.useRealTimers(); } + }, 120_000); + +`; +let status = 1; +try { + for (const [index, kind] of ["query", "work"].entries()) { + const source = git("show", `${baseline}:server/src/services/native-runtime/native-response-presentation-${kind}.ts`); + const materialized = kind === "work" ? source.replace("./native-response-presentation-query.js", "./native-response-presentation-query.baseline.js") : source; + writeFileSync(path.join(root, files[index]), materialized); + writeFileSync(path.join(output, path.basename(files[index])), materialized); + } + writeFileSync(path.join(output, "benchmark-fragment.txt"), benchmark); + writeFileSync(path.join(root, integration), original.replace(anchor, benchmark + anchor)); + const log = openSync(path.join(output, "vitest.log"), "wx"); + try { + const result = spawnSync("pnpm", ["exec", "vitest", "run", "--project", "@paperclipai/server", integration, + "-t", "measures equivalent completed accepted-response work without artificial sleeps"], + { cwd: root, env, stdio: ["ignore", log, log] }); + if (result.error) throw result.error; + status = result.status ?? 1; + } finally { closeSync(log); } + if (status === 0) { + const pairs = readFileSync(path.join(output, "pairs.jsonl"), "utf8").trim().split("\n").map(line => JSON.parse(line)); + assert.equal(pairs.length, 10); + assert.ok(pairs.every(pair => pair.node === "v24.22.2" && pair.completedResponses === 4)); + } +} finally { + writeFileSync(path.join(root, integration), original); + for (const file of files) rmSync(path.join(root, file), { force: true }); +} +console.log(`Benchmark ${status === 0 ? "passed" : "failed"}; artifacts: ${output}`); +process.exitCode = status; diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index eda5fab7109..a139347bed6 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -64125,24 +64125,32 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } }); + type SharedEndpointProgress = { + fixture: Awaited>; + configured: Awaited> | Awaited>; + suffix: string; + }; + async function failedChatRetryFixture( provider: "slack" | "telegram" | "github", externalActorId?: string, linkedOriginalActor = false, githubUnavailableFile = false, currentGitHubPr = false, + shared?: SharedEndpointProgress, ) { // Keep the policy under which the source was admitted. Inserting an // upgraded GitHub policy after receipt correctly invalidates retry proof. const context = await safeNativeProgressFixture( provider, - "91", + shared?.suffix ?? "91", "channel", externalActorId, linkedOriginalActor, githubUnavailableFile, false, currentGitHubPr, + shared, ); const [issue] = await readIssuesById(context.conversation.issueId); const [action] = await db @@ -68625,9 +68633,10 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { githubUnavailableFile = false, upgradedGitHub = true, currentGitHubPr = false, + shared?: SharedEndpointProgress, ) { - const fixture = await seedCompany(); - const configured = + const fixture = shared?.fixture ?? await seedCompany(); + const configured = shared?.configured ?? ( provider === "slack" ? await configuredSlackEndpoint(fixture) : provider === "github" @@ -68641,7 +68650,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ? await configuredDiscordEndpoint(fixture) : provider === "microsoft-teams" ? await configuredTeamsEndpoint(fixture) - : await configuredTelegramEndpoint(fixture); + : await configuredTelegramEndpoint(fixture)); const { callbacks, endpoint, runtime, service } = configured; const telegramChatId = `77113${suffix.padStart(3, "0")}`; const teamsConversationId = `19:safe-progress-${suffix}@thread.${teamsSurface === "channel" ? "tacv2" : "v2"}`; @@ -68769,7 +68778,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { message: sourceMessage, trigger: thread.thread.isDM ? "direct_message" : "mention", }); - const [conversation] = await readChatConversationsByEndpointId(endpoint.id); + const conversations = await readChatConversationsByEndpointId(endpoint.id); + const conversation = shared ? conversations.find(row => row.externalThreadId === thread.thread.id) : conversations[0]; if (!conversation) throw new Error("Expected safe-progress conversation"); const providerRuntime = runtime.endpoints.get(endpoint.id); if (!providerRuntime) throw new Error("Expected safe-progress runtime"); @@ -74594,12 +74604,15 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ) => Promise, summary?: (issueId: string) => string, finalize = true, + shared?: SharedEndpointProgress, ) { const context = await failedChatRetryFixture( provider, provider === "github" ? "42" : undefined, false, githubUnavailableFile, + false, + shared, ); if (provider === "github") { const config = await githubChatManagementService(db).configuration(context.endpoint.id, "owner-user"); @@ -74787,6 +74800,126 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("settles a shared-endpoint accepted-response backlog without retrying its own locks", async () => { + const fixture = await seedCompany(); + const configured = await configuredTelegramEndpoint(fixture); + const contexts = []; + for (let index = 0; index < 4; index++) { + contexts.push(await committedChatResponseRecoveryFixture("telegram", "coordinator", false, + undefined, undefined, true, { fixture, configured, suffix: String(211 + index) })); + } + expect(new Set(contexts.map(context => context.issue.id)).size).toBe(4); + const independent = await committedChatResponseRecoveryFixture("telegram"); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + // Hold one real comment write after authority locks are acquired. Another + // endpoint must still publish while this endpoint's backlog waits. + const functionName = `hold_presentation_${randomUUID().replaceAll("-", "")}`; + const gate = Math.floor(Math.random() * 1_000_000_000); + let release!: () => void, locked!: () => void; + let holderPid = 0; + const hold = new Promise(resolve => { release = resolve; }); + const acquired = new Promise(resolve => { locked = resolve; }); + const holder = db.transaction(async tx => { + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`); + holderPid = Number(backend.pid); + await tx.execute(sql`select pg_advisory_xact_lock(${gate})`); locked(); await hold; + }); + await acquired; + const sharedRuns = contexts.map(context => `'${context.runId}'::uuid`).join(","); + await db.execute(sql.raw(`create function ${functionName}() returns trigger language plpgsql as $$ begin if NEW.created_by_run_id in (${sharedRuns}) then perform pg_advisory_xact_lock(${gate}); end if; return NEW; end $$`)); + await db.execute(sql.raw(`create trigger ${functionName} before insert on issue_comments for each row execute function ${functionName}()`)); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await expect.poll(async () => { + const [state] = await db.execute(sql`select exists (select 1 from pg_stat_activity + where datname = current_database() and ${holderPid} = any(pg_blocking_pids(pid))) as blocked`); + return state.blocked; + }, { timeout: 10_000 }).toBe(true); + await expect.poll(async () => (await db.select({ id: issueComments.id }).from(issueComments) + .where(eq(issueComments.createdByRunId, independent.runId))).length, { timeout: 10_000 }).toBe(1); + expect(await db.select({ id: issueComments.id }).from(issueComments) + .where(inArray(issueComments.createdByRunId, contexts.map(context => context.runId)))).toHaveLength(0); + release(); await holder; + await worker.ready; + await expect.poll(() => idleWorkSnapshot().active, { timeout: 10_000 }).toBe(0); + expect(errors).toEqual([]); + for (const context of contexts) { + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId)); + expect(comments).toHaveLength(1); + expect(comments[0].body).toBe(context.result.summary); + const publications = await readChatPublicationsByCommentId(comments[0].id); + expect(publications).toHaveLength(1); + expect(publications[0]).toMatchObject({ companyId: fixture.companyId, endpointId: configured.endpoint.id, + issueId: context.issue.id, conversationId: context.conversation.id, payload: { text: context.result.summary } }); + } + expect(await readChatPublicationsByEndpointId(configured.endpoint.id)).toHaveLength(4); + expect(coordinator.nextWakeAt()).toBeNull(); + } finally { + release(); await holder; + worker.stop(); await coordinator.stop(); + await db.execute(sql.raw(`drop trigger ${functionName} on issue_comments`)); + await db.execute(sql.raw(`drop function ${functionName}()`)); + await configured.service.shutdown(); await independent.service.shutdown(); + } + }); + + it("settles responses on different endpoints for one principal without retrying its own identity lock", async () => { + const fixture = await seedCompany(); + const first = await configuredSlackEndpoint(fixture); + const second = await configuredSlackEndpoint(fixture); + const contexts = [ + await committedChatResponseRecoveryFixture("slack", "coordinator", false, undefined, undefined, true, + { fixture, configured: first, suffix: "221" }), + await committedChatResponseRecoveryFixture("slack", "coordinator", false, undefined, undefined, true, + { fixture, configured: second, suffix: "222" }), + ]; + expect(contexts[0].action.principalId).toBeTruthy(); + expect(contexts[0].action.principalId).toBe(contexts[1].action.principalId); + expect(contexts[0].endpoint.id).not.toBe(contexts[1].endpoint.id); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const functionName = `hold_identity_${randomUUID().replaceAll("-", "")}`; + await db.execute(sql.raw(`create function ${functionName}() returns trigger language plpgsql as $$ begin perform pg_sleep(0.1); return NEW; end $$`)); + await db.execute(sql.raw(`create trigger ${functionName} before insert on issue_comments for each row execute function ${functionName}()`)); + const worker = nativeResponsePresentationWork(db).register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => idleWorkSnapshot().active, { timeout: 10_000 }).toBe(0); + expect(errors).toEqual([]); + for (const context of contexts) { + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, context.runId)); + expect(comments).toHaveLength(1); + expect(comments[0].body).toBe(context.result.summary); + const publications = await readChatPublicationsByCommentId(comments[0].id); + expect(publications).toHaveLength(1); + expect(publications[0]).toMatchObject({ companyId: fixture.companyId, endpointId: context.endpoint.id, + issueId: context.issue.id, conversationId: context.conversation.id, payload: { text: context.result.summary } }); + } + expect(coordinator.nextWakeAt()).toBeNull(); + } finally { + worker.stop(); await coordinator.stop(); + await db.execute(sql.raw(`drop trigger ${functionName} on issue_comments`)); + await db.execute(sql.raw(`drop function ${functionName}()`)); + await first.service.shutdown(); await second.service.shutdown(); + } + }); + + it("does not treat an accepted-response endpoint scheduling hint as current authority", async () => { + const context = await committedChatResponseRecoveryFixture("telegram"); + const work = nativeResponsePresentationWork(db); + try { + const [entry] = await work.load({ companyId: context.fixture.companyId }); + expect(entry.endpointId).toBe(context.endpoint.id); + await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, context.endpoint.id)); + // A stale or forged hint cannot authorize presentation against the real source. + await expect(work.check({ ...entry, endpointId: randomUUID(), principalId: randomUUID() })).resolves.toBeNull(); + expect(await db.select({ id: issueComments.id }).from(issueComments) + .where(eq(issueComments.createdByRunId, context.runId))).toHaveLength(0); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + } finally { await context.service.shutdown(); } + }); + it("repairs a committed reply and its recovery notice without reversing row locks", async () => { const context = await committedChatResponseRecoveryFixture("telegram"); const scope = { companyId: context.fixture.companyId, issueId: context.issue.id, runId: context.runId }; @@ -76163,7 +76296,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { await context.service.shutdown(); } }); - it("leaves a contended accepted response retryable without reclassifying or partially publishing it", async () => { + it.each(["action", "endpoint"] as const)("leaves a contended accepted response retryable without reclassifying or partially publishing it (%s)", async target => { const context = await committedChatResponseRecoveryFixture("telegram"); let release!: () => void; let observed!: () => void; @@ -76174,11 +76307,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { observed = resolve; }); const holder = db.transaction(async (tx) => { - await tx - .select() - .from(chatActions) - .where(eq(chatActions.id, context.action.id)) - .for("update"); + if (target === "endpoint") { + await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, context.endpoint.id)).for("no key update"); + } else { + await tx.select().from(chatActions).where(eq(chatActions.id, context.action.id)).for("update"); + } observed(); await hold; }); diff --git a/server/src/__tests__/scoped-deadline-work.test.ts b/server/src/__tests__/scoped-deadline-work.test.ts index 84cd4d77374..be1b568cfcd 100644 --- a/server/src/__tests__/scoped-deadline-work.test.ts +++ b/server/src/__tests__/scoped-deadline-work.test.ts @@ -40,7 +40,7 @@ it("preserves a subsecond replacement deadline when the owner does not require p type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; -function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) { +function concurrentFixture(entries: Entry[], limit = 4, concurrent = true, key?: (entry: Entry) => string | readonly string[]) { vi.useFakeTimers(); const onError = vi.fn(); const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) }); @@ -66,7 +66,7 @@ function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) { coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, onError }); const options = { queue: DELIVERY_QUEUES.nativeResponsePresentation, - concurrentChecks: concurrent ? { limit, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` } : undefined, + concurrentChecks: concurrent ? { limit, key: key ?? ((entry: Entry) => `${entry.companyId}:${entry.issueId}`) } : undefined, decodeScope: () => undefined, matches: () => true, load, @@ -85,6 +85,88 @@ function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) { const entry = (id: string, issueId = id, companyId = "company"): Entry => ({ id, issueId, companyId, dueAt: 0 }); +it("merges transitive resource conflicts in candidate order without blocking independent work", async () => { + const keys: Record = { + "1": ["issue:a", "endpoint:first", "endpoint:first"], + "2": ["issue:b", "endpoint:second"], + "3": ["issue:a", "endpoint:second"], + "4": ["issue:c", "endpoint:independent"], + "5": ["issue:without-endpoint"], + "6": ["issue:without-endpoint"], + }; + const f = concurrentFixture([1, 2, 3, 4, 5, 6].map(id => entry(String(id))), 4, true, entry => keys[entry.id]); + try { + await vi.waitFor(() => expect(f.started).toEqual(["1", "4", "5"]), { timeout: 100 }); + f.release("4"); f.release("5"); + await vi.waitFor(() => expect(f.started).toContain("6"), { timeout: 100 }); + expect(f.started).not.toContain("2"); + f.release("1"); + await vi.waitFor(() => expect(f.started).toContain("2"), { timeout: 100 }); + expect(f.started).not.toContain("3"); + f.release("2"); + await vi.waitFor(() => expect(f.started).toContain("3"), { timeout: 100 }); + expect(f.started.filter(id => ["1", "2", "3"].includes(id))).toEqual(["1", "2", "3"]); + } finally { f.releaseAll(); await f.worker.ready; } + expect(f.onError).not.toHaveBeenCalled(); + expect(coordinator!.nextWakeAt()).toBeNull(); +}); + +it("retains unstarted resource siblings through drain and resumes without reconstruction", async () => { + const f = concurrentFixture([entry("1"), entry("2"), entry("3")], 4, true, + entry => [entry.issueId, ...(entry.id === "3" ? ["independent"] : ["shared-endpoint"])]); + try { + await vi.waitFor(() => expect(f.started).toEqual(["1", "3"]), { timeout: 100 }); + f.pause(); f.releaseAll(); await f.worker.ready; + expect(f.started).not.toContain("2"); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.started).not.toContain("2"); + f.resume(); await vi.advanceTimersByTimeAsync(1); + expect(f.started).toEqual(["1", "3", "2"]); + expect(f.load).toHaveBeenCalledOnce(); + expect(f.onError).not.toHaveBeenCalled(); + } finally { f.releaseAll(); await f.worker.ready; } +}); + +it("retains an in-flight rescope and admits a bridging newcomer once after active groups settle", async () => { + const keys: Record = { + "1": ["issue:a", "endpoint:first"], "2": ["issue:b", "endpoint:second"], + "3": ["issue:a", "endpoint:second"], + }; + const f = concurrentFixture([entry("1"), entry("2")], 4, true, entry => keys[entry.id]); + try { + await vi.waitFor(() => expect(f.started).toEqual(["1", "2"]), { timeout: 100 }); + f.entries.push(entry("3")); f.worker.invalidate("new-scope"); + f.release("1"); + await vi.waitFor(() => expect(f.active.has("1")).toBe(false), { timeout: 100 }); + expect(f.load).toHaveBeenCalledOnce(); + expect(f.started).not.toContain("3"); + f.release("2"); await f.worker.ready; + await vi.advanceTimersByTimeAsync(1000); + expect(f.load).toHaveBeenCalledTimes(2); + expect(f.started).toEqual(["1", "2", "3"]); + } finally { f.releaseAll(); await vi.advanceTimersByTimeAsync(1); await f.worker.ready; } + expect(f.started.filter(id => id === "3")).toHaveLength(1); + expect(f.onError).not.toHaveBeenCalled(); +}); + +it("serializes a shared principal across endpoints while unrelated resources progress", async () => { + const keys: Record = { + "1": ["issue:a", "endpoint:a", "principal:shared"], + "2": ["issue:b", "endpoint:b", "principal:shared"], + "3": ["issue:c", "endpoint:c", "principal:independent"], + }; + const f = concurrentFixture([entry("1"), entry("2"), entry("3")], 4, true, entry => keys[entry.id]); + try { + await vi.waitFor(() => expect(f.started).toEqual(["1", "3"]), { timeout: 100 }); + f.release("3"); + await vi.waitFor(() => expect(f.active.has("3")).toBe(false), { timeout: 100 }); + expect(f.started).not.toContain("2"); + f.release("1"); + await vi.waitFor(() => expect(f.started).toEqual(["1", "3", "2"]), { timeout: 100 }); + } finally { f.releaseAll(); await f.worker.ready; } + expect(f.onError).not.toHaveBeenCalled(); +}); + it("checks independent tasks concurrently with a bounded number of active checks", async () => { const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index)))); try { @@ -157,22 +239,23 @@ it("retains unstarted checks through drain and resumes without idle discovery re expect(f.onError).not.toHaveBeenCalled(); }); -it("stops starting checks and awaits every admitted check during shutdown", async () => { - const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index)))); +it.each(["task", "resources"] as const)("stops starting checks and awaits every admitted %s check during shutdown", async mode => { + const f = concurrentFixture(Array.from({ length: 6 }, (_, index) => entry(String(index))), 4, true, + mode === "resources" ? entry => [entry.issueId, `endpoint:${Number(entry.id) % 2}`] : undefined); let stopped = false; let stop: Promise | undefined; try { - await vi.waitFor(() => expect(f.started).toHaveLength(4), { timeout: 100 }); + await vi.waitFor(() => expect(f.started).toHaveLength(mode === "resources" ? 2 : 4), { timeout: 100 }); stop = coordinator!.stop().then(() => { stopped = true; }); expect(f.signals.every(signal => signal.aborted)).toBe(true); expect(stopped).toBe(false); f.release("0"); await vi.waitFor(() => expect(f.active.has("0")).toBe(false), { timeout: 100 }); expect(stopped).toBe(false); - expect(f.started).toHaveLength(4); + expect(f.started).toHaveLength(mode === "resources" ? 2 : 4); } finally { f.releaseAll(); await f.worker.ready; await stop; } expect(stopped).toBe(true); - expect(f.started).toHaveLength(4); + expect(f.started).toHaveLength(mode === "resources" ? 2 : 4); }); it("keeps failures isolated and retries their deadline without repeating successful checks", async () => { diff --git a/server/src/services/native-runtime/native-response-presentation-query.ts b/server/src/services/native-runtime/native-response-presentation-query.ts index d181e3dc9c2..bb2bbf129c7 100644 --- a/server/src/services/native-runtime/native-response-presentation-query.ts +++ b/server/src/services/native-runtime/native-response-presentation-query.ts @@ -1,15 +1,18 @@ import { QueryBuilder } from "drizzle-orm/pg-core"; import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; -import { heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db"; +import { chatActions, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db"; import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js"; export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) { const db = new QueryBuilder(); return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + endpointId: chatActions.endpointId, principalId: chatActions.principalId, }).from(nativeRunFinalizations) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .leftJoin(chatActions, and(eq(chatActions.id, heartbeatRuns.wakeupRequestId), + eq(chatActions.companyId, heartbeatRuns.companyId))) .innerJoin(issues, and(eq(issues.id, nativeRunFinalizations.issueId), eq(issues.companyId, nativeRunFinalizations.companyId))) .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index 25a2c65a4dd..50c89162c32 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -14,7 +14,7 @@ export function registerNativeResponsePresentationWork(db: Db, return nativeResponsePresentationWork(db).register(coordinator, canRun); } -type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; +type Entry = { id: string; companyId: string; issueId: string; agentId: string; endpointId?: string | null; principalId?: string | null; dueAt: number }; export function nativeResponsePresentationWork(db: Db) { const candidates = (scope?: NativeResponsePresentationScope, after?: string, runId?: string) => { @@ -31,7 +31,7 @@ export function nativeResponsePresentationWork(db: Db) { let after: string | undefined; while (canRun() && !signal?.aborted) { const rows = await measureRunPhase("response.resolve", () => candidates(scope, after)); - for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, endpointId: row.endpointId, principalId: row.principalId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); if (rows.length < 100) return entries; after = rows[rows.length - 1].id; } @@ -56,7 +56,13 @@ export function nativeResponsePresentationWork(db: Db) { function register(coordinator: ReturnType, canRun: () => boolean) { const worker = registerScopedDeadlineWork(coordinator, { queue: DELIVERY_QUEUES.nativeResponsePresentation, - concurrentChecks: { limit: 4, key: (entry: Entry) => `${entry.companyId}:${entry.issueId}` }, + // Endpoint and principal IDs are only dispatch hints. Recovery still locks and checks + // the complete current source authority inside its own transaction. + concurrentChecks: { limit: 4, key: (entry: Entry) => [ + `issue:${entry.companyId}:${entry.issueId}`, + ...(entry.endpointId ? [`endpoint:${entry.companyId}:${entry.endpointId}`] : []), + ...(entry.principalId ? [`principal:${entry.companyId}:${entry.principalId}`] : []), + ] }, decodeScope: (hint: string): NativeResponsePresentationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), matches: (entry: Entry, scope?: NativeResponsePresentationScope) => !scope || entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) && (!scope.agentId || entry.agentId === scope.agentId), load: (scope, signal) => load(scope, signal, canRun), diff --git a/server/src/services/scoped-deadline-work.ts b/server/src/services/scoped-deadline-work.ts index bfcf13a2e17..cdd69a6801e 100644 --- a/server/src/services/scoped-deadline-work.ts +++ b/server/src/services/scoped-deadline-work.ts @@ -12,7 +12,7 @@ export function registerScopedDeadlineWork( queue: DeliveryQueue; batchSize?: number; minimumPassIntervalMs?: number; - concurrentChecks?: { limit: number; key(entry: Entry): string }; + concurrentChecks?: { limit: number; key(entry: Entry): string | readonly string[] }; reconstructionIntervalMs?: number | ((scope: Scope | undefined) => number); beforePass?(): Promise; decodeScope(hint: string): Scope | undefined; @@ -116,9 +116,27 @@ export function registerScopedDeadlineWork( && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)); const batch = takeCompanyWorkBatch(due, options.batchSize ?? 25, lastCompany); lastCompany = batch.lastCompany; - const groups = new Map(); - for (const entry of batch.entries) { - const key = options.concurrentChecks?.key(entry) ?? ""; + // An owner may share more than one resource (for example a task and an + // endpoint). Merge intersecting keys before dispatch so waiting siblings + // do not occupy the bounded check slots ahead of independent work. + const parents = batch.entries.map((_, index) => index); + const root = (index: number): number => { + while (parents[index] !== index) index = parents[index]; + return index; + }; + const owners = new Map(); + for (const [index, entry] of batch.entries.entries()) { + const configured = options.concurrentChecks?.key(entry) ?? ""; + const keys = typeof configured === "string" ? [configured] : configured; + for (const key of keys) { + const previous = owners.get(key); + if (previous === undefined) owners.set(key, index); + else parents[root(index)] = root(previous); + } + } + const groups = new Map(); + for (const [index, entry] of batch.entries.entries()) { + const key = root(index); const group = groups.get(key) ?? []; group.push(entry); groups.set(key, group); }