fix(server): deliver answers promptly during web handoff (#18530)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Its web servers can overlap while Railway replaces one deployment.
> - An answer can reach the replacement while the original server owns
the native execution.
> - A competing delivery claim can make the owner miss the first
notification.
> - The normal retry interval then delays the answer by 30 seconds.
> - This pull request gives pending answers a one-second retry in
ephemeral mode and verifies delivery between separate processes.

## Linked Issues or Issue Description

Refs #18518 and #18526. This closes a measured native-question delivery
delay during control-plane web deployment overlap.

## What Changed

- Register question delivery through its owning service. Use a
one-second retry for ephemeral servers and retain the configured cadence
for ordinary servers.
- Keep the existing durable claims, response identity, authorization and
continuation behavior.
- Add a two-process PostgreSQL regression for a claim held by the
non-owning server. Verify one command, settlement, idle behavior and
reactivation by another committed answer.
- Keep the assembled startup fixture on the real registration function.
Update the deployment plan with current evidence and remaining
acceptance work.

## Verification

- The same local two-process race took 30.157 seconds before the fix and
1.197 seconds after it. The command sink is simulated; PostgreSQL,
notification transport, interaction response, delivery claim and worker
are real.
- The cross-process regression and assembled startup suite passed 122
tests. No further sweeps or pending inspections occur after settlement.
A second answer wakes the owner through PostgreSQL without a test-owned
sweep.
- Server TypeScript compilation passed. All PR CI gates passed on
d1d426c0dc, including full tests, typecheck, build, browser shards and
serialized suites. Greptile reviewed that head at 5/5 with no findings.
- Live provider continuation, hosted staging deployments and production
acceptance are still separate work.

## Risks

- An unresolved answer causes more frequent reconciliation in ephemeral
mode. Settled history does not keep the worker active. Existing claim
fencing and bounded error attempts are unchanged.
- A 30-second abandoned claim still needs its existing stale-claim
recovery. This change addresses released claims, not proof of provider
liveness.
- The test simulates the provider command sink. It cannot establish that
a paid model consumed the answer or that a remote execution survived a
deploy.
- No schema, API or UI changes.

## Model Used

OpenAI Codex, GPT-6. Tool use, code execution and repository review. The
runtime does not expose a more specific model ID or context-window size.

## 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
- [x] 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-11-01 03:39:48 -06:00
1 parent 16787e4167
commit 73cd8eb9a5
6 files changed
+431 -3

No files matched your search

@@ -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"
});
}
});
@@ -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<Message>((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<void>(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);
@@ -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<typeof import("../services/question-response-delivery.js")>(),
questionResponseDeliveryService: vi.fn(() => ({
hasPending: vi.fn(async () => false),
sweepPending: vi.fn(async () => ({
+3 -2
View File
@@ -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<typeof import("./services/tool-action-delivery.js").toolActionDeliveryService>],
] 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 => {
@@ -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<typeof createDeliveryWorkCoordinator>,
service: Pick<ReturnType<typeof questionResponseDeliveryService>, "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";