From d1d426c0dc60e3fa963ef833c96ccbc6f8976eaf Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 1 Nov 2026 03:16:40 -0600 Subject: [PATCH] fix(server): deliver answers promptly during web handoff Reduce pending-answer retry latency only for ephemeral servers. Verify real PostgreSQL delivery between two processes, one command per answer, quiet settlement, and reactivation by a new answer. Co-Authored-By: Paperclip --- .../2026-10-31-zero-downtime-web-deploys.md | 35 +++ .../helpers/native-question-owner.mjs | 93 ++++++ .../native-question-process-handoff.test.ts | 283 ++++++++++++++++++ .../server-startup-feedback-export.test.ts | 3 +- server/src/index.ts | 5 +- .../services/question-response-delivery.ts | 15 + 6 files changed, 431 insertions(+), 3 deletions(-) create mode 100644 server/src/__tests__/helpers/native-question-owner.mjs create mode 100644 server/src/__tests__/native-question-process-handoff.test.ts diff --git a/doc/plans/2026-10-31-zero-downtime-web-deploys.md b/doc/plans/2026-10-31-zero-downtime-web-deploys.md index 878a1c05d94..8d30b2bc869 100644 --- a/doc/plans/2026-10-31-zero-downtime-web-deploys.md +++ b/doc/plans/2026-10-31-zero-downtime-web-deploys.md @@ -357,3 +357,38 @@ then detach only after a matching receipt and confirmed process exit. Remaining acceptance is unchanged: live Stop/answer/follow-up and continuation, packaged fresh/migrated overlap, ten staging deployments including rollback, repair/sleep/wake and idle suspension, then one internal production canary. + + +### Current delivery state, 2026-11-01 09:15 UTC + +Qualification #18526 merged as `16787e4167fa94bd968419cf6e6e9e8ebb2748c4` +with all checks green and a 5/5 review. Cloud enrollment #1098 merged as +`316163b28d4f9e811e810a3226ea46b33a273ce7` after 5/5 review and green CI. +Its review fixes explicitly reopen registry routing on completion and allow +same-attempt final-boot retry without reattaching the retained volume. No hosted +stack has yet been enrolled by this chat. + +Core enrollment #18529 is at `6d31d1265d`. Twenty focused checks passed, +including a rebuilt plain-Node executable and real PostgreSQL/S3. Owner checks +now block unresolved local working copies even for terminal runs, drain all +expired trace pages, and expire successful key-export authority at the original +deadline. An integration fixture initially failed cleanup because trace expiry +writes activity rows; cleanup now deletes those before their runs. The corrected +run passed. Initial CI failures were a workspace-runtime health probe and an +extra select in a run-dispatch idle assertion; latest-head CI/review is pending. + +An independent two-process question-answer probe exposed a real retry delay: +B releases its claim after A inspects it, leaving the owner waiting 30.157s. +The production pending-answer worker now uses a one-second retry in ephemeral +mode; the same probe delivered one command in 1.197s. Ordinary server cadence +is retained. A committed regression is in preparation: real PostgreSQL and two +processes, with only the provider command sink simulated. It checks no repeated +sweeps/inspections after settlement and reactivation by the next committed answer. +The regression and assembled startup suite passed 122 tests. This does not +prove live provider response or active remote execution survival. + +The refreshed E2E OpenAI credential remains outstanding. No subsequent paid +attempt has been made. Remaining acceptance still includes live Stop/answer/ +follow-up/continuation, packaged overlap with migrated data and large cold +restoration, ten staging deployments including rollback, repair/sleep/wake and +idle suspension, and one internal production canary forward/rollback. diff --git a/server/src/__tests__/helpers/native-question-owner.mjs b/server/src/__tests__/helpers/native-question-owner.mjs new file mode 100644 index 00000000000..59118a78b18 --- /dev/null +++ b/server/src/__tests__/helpers/native-question-owner.mjs @@ -0,0 +1,93 @@ +// Separate process: only the provider command sink is simulated. +import { createDb } from "@paperclipai/db"; +import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.ts"; +import { deploymentNotifications } from "../../services/deployment-notifications.ts"; +import { questionResponseDeliveryService, registerQuestionResponseDeliveryWork } from "../../services/question-response-delivery.ts"; +import { deliverNativeQuestionResponse, registerNativeQuestionCommandTarget } from "../../services/native-runtime/native-question-bridge.ts"; +let db, + coordinator, + bridge, + worker, + release, + sweeps = 0, + inspections = 0; +process.on("message", async message => { + try { + if (message.type === "init") { + db = createDb(message.url); + coordinator = createDeliveryWorkCoordinator({ + owner: db, + canRun: () => true, + onError: () => process.send({ + type: "error" + }) + }); + bridge = deploymentNotifications(db, coordinator, { + directDatabaseUrl: message.url + }); + await bridge.ready(); + release = registerNativeQuestionCommandTarget({ + binding: message.binding, + queueCommand: (type, payload, commandId) => { + process.send({ + type: "command", + commandType: type, + requestId: commandId, + payload, + commandId + }); + return { + commandId, + controllerSeq: 1 + }; + } + }); + process.send({ + type: "ready" + }); + } else if (message.type === "start-worker") { + const service = questionResponseDeliveryService(db, { + heartbeat: { + wakeup: async () => { + throw new Error("unexpected fallback"); + } + }, + resolveNativeQuestion: interaction => deliverNativeQuestionResponse(db, interaction) + }); + worker = registerQuestionResponseDeliveryWork(coordinator, { + sweepPending: () => { + sweeps++; + return service.sweepPending(); + }, + hasPending: () => { + inspections++; + return service.hasPending(); + } + }); + await worker.ready; + process.send({ + type: "worker-ready" + }); + } else if (message.type === "snapshot") { + process.send({ + type: "snapshot", + requestId: message.requestId, + sweeps, + inspections + }); + } else if (message.type === "stop") { + release?.(); + await worker?.stop(); + await bridge?.stop(); + await coordinator?.stop(); + await db?.$client.end({ + timeout: 1 + }); + process.exit(0); + } + } catch { + process.send({ + type: "error" + }); + } +}); diff --git a/server/src/__tests__/native-question-process-handoff.test.ts b/server/src/__tests__/native-question-process-handoff.test.ts new file mode 100644 index 00000000000..f73f275e158 --- /dev/null +++ b/server/src/__tests__/native-question-process-handoff.test.ts @@ -0,0 +1,283 @@ +import { expect, it } from "vitest"; +import { setTimeout as delay } from "node:timers/promises"; +import { fork } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { createDb, companies, agents, issues, heartbeatRuns, authUsers, companyMemberships, issueQuestionResponseDeliveries } from "@paperclipai/db"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { questionResponseDeliveryService } from "../services/question-response-delivery.js"; +import { deliverNativeQuestionResponse, projectNativeRuntimeRequest } from "../services/native-runtime/native-question-bridge.js"; +import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { deploymentNotifications } from "../services/deployment-notifications.js"; +it("delivers a B-to-A native answer promptly after a competing claim and becomes idle after settlement", async () => { + const database = await startEmbeddedPostgresTestDatabase("question-handoff-"); + const db = createDb(database.connectionString), + companyId = randomUUID(), + agentId = randomUUID(), + issueId = randomUUID(), + runId = randomUUID(), + userId = randomUUID(), + sessionId = randomUUID(), + runnerId = randomUUID(); + const child = fork(new URL("./helpers/native-question-owner.mjs", import.meta.url), [], { + execArgv: ["--import", import.meta.resolve("tsx")], + stdio: ["ignore", "ignore", "ignore", "ipc"] + }); + type Message = { + type: string; + commandType?: string; + commandId?: string; + payload?: { + response: unknown; + }; + requestId?: string; + sweeps?: number; + inspections?: number; + }; + const messages: Message[] = []; + const waiters = new Set<() => void>(); + child.on("message", value => { + const message = value as Message; + messages.push(message); + for (const check of waiters) check(); + }); + function wait(type: string, timeout = 45000, requestId?: string) { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + waiters.delete(check); + reject(new Error(`Timed out waiting for ${type}`)); + }, timeout); + function check() { + const m = messages.find(m => m.type === type && (!requestId || m.requestId === requestId) || m.type === "error"); + if (m) { + clearTimeout(timer); + waiters.delete(check); + m.type === "error" ? reject(new Error("Owner process failed")) : resolve(m); + } + } + waiters.add(check); + check(); + }); + } + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ + owner: db, + canRun: () => true, + onError: error => errors.push(error) + }), + bridge = deploymentNotifications(db, coordinator, { + directDatabaseUrl: database.connectionString + }); + try { + await bridge.ready(); + await db.insert(companies).values({ + id: companyId, + name: "Question handoff proof", + issuePrefix: "QHO" + }); + await db.insert(authUsers).values({ + id: userId, + name: "Operator", + email: `${userId}@example.test`, + createdAt: new Date(), + updatedAt: new Date() + }); + await db.insert(companyMemberships).values({ + companyId, + principalType: "user", + principalId: userId, + status: "active", + membershipRole: "member" + }); + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Handoff fixture", + adapterType: "paperclip_runner", + status: "running", + adapterConfig: { + provider: "codex" + } + }); + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Answer across processes", + status: "in_progress", + assigneeAgentId: agentId, + responsibleUserId: userId + }); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + status: "running", + runtimeMode: "native", + nativeIssueId: issueId, + nativeSessionId: sessionId, + runnerInstanceId: runnerId, + driverKind: "codex", + contextSnapshot: { + issueId + } + }); + const createQuestion = (requestId: string) => projectNativeRuntimeRequest({ + db, + binding: { + companyId, + issueId, + runId, + agentId, + normalizedSessionId: sessionId, + runnerSourceInstanceId: runnerId + }, + event: { + schema: "paperclip.prp.event.v1", + sourceEventId: requestId, + sourceSeq: 1, + sourceInstanceId: runnerId, + sourceKind: "runner", + runId, + normalizedSessionId: sessionId, + eventType: "runtime_request.created", + schemaVersion: 1, + priority: 0, + emittedAt: new Date().toISOString(), + payload: { + request: { + schema: "paperclip.runtime_request.v2", + requestKind: "runtime", + requestId, + type: "input", + status: "pending", + input: { + schema: "paperclip.question_set.v1", + questions: [{ + id: "color", + prompt: "Which color?", + required: true, + answerMode: "single_select", + options: [{ + id: "green", + label: "Green" + }] + }] + } + } + } + } + }); + const interaction = await createQuestion("question-1"); + child.send({ + type: "init", + url: database.connectionString, + binding: { + companyId, + issueId, + runId, + agentId + } + }); + await wait("ready"); + await issueThreadInteractionService(db).answerQuestions({ + id: issueId, + companyId, + status: "in_progress" + }, interaction!.id, { + answers: [{ + questionId: "color", + optionIds: ["green"] + }] + }, { + userId + }); + const service = questionResponseDeliveryService(db, { + heartbeat: { + wakeup: async () => { + throw new Error("Unexpected fallback"); + }, + cancelRun: async () => { + throw new Error("Unexpected cancellation"); + } + }, + resolveNativeQuestion: async interaction => { + // B has claimed the delivery. A must observe that claim before B + // discovers it lacks the native target and releases the row for retry. + child.send({ + type: "start-worker" + }); + await wait("worker-ready"); + return deliverNativeQuestionResponse(db, interaction); + } + }); + const started = Date.now(); + const immediate = await service.deliver(interaction!.id); + const command = await wait("command", 5000); + expect(immediate).toBeNull(); + expect(Date.now() - started).toBeLessThan(5000); + expect(command).toMatchObject({ + commandType: "request.resolve", + commandId: `question_${interaction!.id}`, + payload: { + response: { + schema: "paperclip.question_response.v1", + answers: { + color: { + selectedOptionIds: ["green"] + } + } + } + } + }); + await expect.poll(async () => (await db.select().from(issueQuestionResponseDeliveries))[0]?.status).toBe("delivered"); + const snapshot = async () => { + const requestId = randomUUID(); + child.send({ + type: "snapshot", + requestId + }); + return wait("snapshot", 5000, requestId); + }; + await delay(1500); + const settled = await snapshot(); + await delay(2500); + const idle = await snapshot(); + expect({ + sweeps: idle.sweeps, + inspections: idle.inspections + }).toEqual({ + sweeps: settled.sweeps, + inspections: settled.inspections + }); + expect(messages.filter(m => m.type === "command")).toHaveLength(1); + // A has become idle. The next real producer on B must wake it through + // PostgreSQL without a test-owned sweep or another child control message. + const next = await createQuestion("question-2"); + await issueThreadInteractionService(db).answerQuestions({ id: issueId, companyId, status: "in_progress" }, next!.id, + { answers: [{ questionId: "color", optionIds: ["green"] }] }, { userId }); + expect(await wait("command", 5_000, `question_${next!.id}`)).toMatchObject({ commandType: "request.resolve" }); + await expect.poll(async () => (await db.select().from(issueQuestionResponseDeliveries)).every(row => row.status === "delivered")).toBe(true); + expect(messages.filter(m => m.type === "command")).toHaveLength(2); + expect(errors).toEqual([]); + } finally { + if (child.connected) child.send({ + type: "stop" + }); + await new Promise(resolve => { + if (child.exitCode !== null || child.signalCode !== null) return resolve(); + const timer = setTimeout(() => { + child.kill("SIGKILL"); + }, 10000); + child.once("exit", () => { + clearTimeout(timer); + resolve(); + }); + }); + await bridge.stop(); + await coordinator.stop(); + await db.$client.end({ + timeout: 1 + }); + await database.cleanup(); + } +}, 90000); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index afbcbbab33a..fbcff811fed 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -537,7 +537,8 @@ vi.mock("../services/connection-intent-delivery.js", () => ({ })), })); -vi.mock("../services/question-response-delivery.js", () => ({ +vi.mock("../services/question-response-delivery.js", async importOriginal => ({ + ...await importOriginal(), questionResponseDeliveryService: vi.fn(() => ({ hasPending: vi.fn(async () => false), sweepPending: vi.fn(async () => ({ diff --git a/server/src/index.ts b/server/src/index.ts index dfea5d79693..ffc3b2d9009 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -156,7 +156,7 @@ import { toolAccessService, workspaceOperationService, } from "./services/index.js"; -import { questionResponseDeliveryService } from "./services/question-response-delivery.js"; +import { questionResponseDeliveryService, registerQuestionResponseDeliveryWork } from "./services/question-response-delivery.js"; import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js"; import { registerStatusCardEvaluationWork, statusCardEvaluationService } from "./services/status-card-evaluation.js"; import { createSecretProposalsService } from "./services/secret-proposals.js"; @@ -1412,7 +1412,6 @@ async function startServerWithDatabaseTeardown( const deliveryWorkers = ([ [DELIVERY_QUEUES.chatCompletion, chatCompletionDeliveries], [DELIVERY_QUEUES.connection, connectionDeliveries], - [DELIVERY_QUEUES.question, questionResponseDeliveries], [DELIVERY_QUEUES.toolAction, app.locals.toolActionDeliveries as ReturnType], ] as const).map(([queue, service]) => deliveryWork.register(queue, { retryMs: config.heartbeatSchedulerIntervalMs, @@ -1420,6 +1419,8 @@ async function startServerWithDatabaseTeardown( hasPending: () => service.hasPending(), ...(queue === DELIVERY_QUEUES.connection ? { nextRunAt: connectionDeliveries.nextRunAt, pausePassiveMaintenance: true } : {}), })); + deliveryWorkers.push(registerQuestionResponseDeliveryWork(deliveryWork, questionResponseDeliveries, + deploymentListenUrl ? undefined : config.heartbeatSchedulerIntervalMs)); // Preserve the existing task-scoped activity fast path. Claims in the // delivery service arbitrate with the coordinator's startup/recovery sweep. const unsubscribeChatCompletions = subscribeAllCompanyLiveEvents(event => { diff --git a/server/src/services/question-response-delivery.ts b/server/src/services/question-response-delivery.ts index 6e85adee7ce..9e45cb41c90 100644 --- a/server/src/services/question-response-delivery.ts +++ b/server/src/services/question-response-delivery.ts @@ -23,10 +23,25 @@ import { logger } from "../middleware/logger.js"; import { logActivity } from "./activity-log.js"; import type { heartbeatService } from "./heartbeat.js"; import { nativeSha256 } from "./native-runtime/canonical.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; const DELIVERY_CLAIM_STALE_MS = 30_000; const DELIVERY_CLAIM_REFRESH_MS = 10_000; const MAX_DELIVERY_ATTEMPTS = 5; + +/** A non-owning controller can release a claim just after the owner observes + * it. Retry that durable obligation promptly during handoff. hasPending keeps + * the worker quiet once all answers settle; retained history is not work. */ +export function registerQuestionResponseDeliveryWork( + coordinator: ReturnType, + service: Pick, "sweepPending" | "hasPending">, + retryMs = 1_000, +) { + return coordinator.register(DELIVERY_QUEUES.question, { + retryMs, run: () => service.sweepPending(), hasPending: () => service.hasPending(), + }); +} const DELIVERY_CORRELATION_PREFIX = "question-response:"; const QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT = "agent_wakeup_requests_question_response_delivery_idempotency_uq";