chore(runner): preserve integration work before PR split

This commit is contained in:
Dotta committed 2026-08-24 08:19:02 -05:00
1 parent 761b60b42c
commit e3df307545
101 files changed
+5865 -395

No files matched your search

@@ -222,7 +222,14 @@ describe("P6-19 native interaction bridge", () => {
kind: "ask_user_questions",
response: {
status: "answered",
result: { version: 1, answers: [{ questionId: "choice", optionIds: ["safe"] }] },
result: {
version: 1,
answers: [{ questionId: "choice", optionIds: ["safe"] }],
summaryMarkdown: [
"Resolved questions and answers:",
"- Which path?: Safe",
].join("\n"),
},
},
},
{
+18 -1
View File
@@ -45,8 +45,9 @@ describe("redaction", () => {
});
it("redacts jwt-looking values even when key name is not sensitive", () => {
const jwt = "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxMjM0NTY3ODkwIn0.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c";
const input = {
session: "aaa.bbb.ccc",
session: jwt,
normal: "plain",
};
@@ -56,6 +57,22 @@ describe("redaction", () => {
expect(result.normal).toBe("plain");
});
it("preserves Paperclip protocol schema identifiers", () => {
expect(sanitizeRecord({
schema: "paperclip.question_set.v1",
nested: {
schema: "paperclip.question_response.v1",
runtimeSchema: "paperclip.runtime_request.v2",
},
})).toEqual({
schema: "paperclip.question_set.v1",
nested: {
schema: "paperclip.question_response.v1",
runtimeSchema: "paperclip.runtime_request.v2",
},
});
});
it("redacts payload objects while preserving null", () => {
expect(redactEventPayload(null)).toBeNull();
expect(redactEventPayload({ password: "hunter2", safe: "value" })).toEqual({
+20
View File
@@ -59,6 +59,7 @@ import {
issueThreadInteractionService,
issueService,
instanceSettingsService,
questionResponseDeliveryService,
reconcileBuiltInAgentsOnStartup,
reconcileCodexLocalManagedHomesOnStartup,
reconcilePersistedRuntimeServicesOnStartup,
@@ -1114,6 +1115,9 @@ export async function startServer(): Promise<StartedServer> {
const ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS = 5 * 60 * 1000;
const environmentLeaseCleanupHeartbeat =
heartbeat ?? heartbeatService(db as any, { pluginWorkerManager });
const questionResponseDeliveries = questionResponseDeliveryService(db as any, {
heartbeat: environmentLeaseCleanupHeartbeat,
});
const runEnvironmentLeaseCleanupSweep = (backoffMs: number) =>
environmentLeaseCleanupHeartbeat
.sweepPendingCleanupLeases({ backoffMs })
@@ -1130,6 +1134,14 @@ export async function startServer(): Promise<StartedServer> {
trackHeartbeatSchedulerWork(runEnvironmentLeaseCleanupSweep(ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS));
};
await questionResponseDeliveries.sweepPending().then((result) => {
if (result.scanned > 0) {
logger.info(result, "startup question-response delivery sweep completed");
}
}).catch((err) => {
logger.error({ err }, "startup question-response delivery sweep failed");
});
if (heartbeat) {
const secretProposals = createSecretProposalsService(db as any);
const decisionExecutor = decisionService(db as any, decisionServiceOptions);
@@ -1540,6 +1552,14 @@ export async function startServer(): Promise<StartedServer> {
logger.error({ err }, "periodic secret proposal expiry sweep failed");
}));
trackHeartbeatSchedulerWork(questionResponseDeliveries.sweepPending()
.then((result) => {
if (result.scanned > 0) logger.info(result, "periodic question-response delivery sweep completed");
})
.catch((err) => {
logger.error({ err }, "periodic question-response delivery sweep failed");
}));
if (heartbeatSchedulerStopped) return;
if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) {
// Periodically reap orphaned runs (5-min staleness threshold) and make sure
+6 -1
View File
@@ -12,7 +12,12 @@ const AUDIT_SURFACE_PAYLOAD_KEY_RE = /^surface$/;
const COMMAND_PAYLOAD_KEY_RE =
/(^command$|^cmd$|command[-_]?line|resolved[-_]?command|PAPERCLIP_RESOLVED_COMMAND)/i;
const COMMAND_ARGS_PAYLOAD_KEY_RE = /^(commandArgs|command_?args|argv)$/i;
const JWT_VALUE_RE = /^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]+)?$/;
// A protocol identifier such as `paperclip.question_set.v1` has the same
// broad three-segment shape as a JWT. Real JSON web tokens encode JSON in the
// first two segments (and therefore conventionally begin with `eyJ`) and have
// a non-trivial signature. Keep the value guard specific enough that durable
// protocol discriminators survive sanitization and replay.
const JWT_VALUE_RE = /^eyJ[A-Za-z0-9_-]{4,}\.eyJ[A-Za-z0-9_-]{4,}\.[A-Za-z0-9_-]{8,}(?:\.[A-Za-z0-9_-]{8,})?$/;
const CLI_SECRET_FLAG_RE = new RegExp(String.raw`^-{1,2}${SECRET_FIELD_NAME_PATTERN}$`, "i");
const JSON_SECRET_FIELD_TEXT_RE = new RegExp(
String.raw`((?:"|')?${SECRET_FIELD_NAME_PATTERN}(?:"|')?\s*:\s*(?:"|'))[^"'` + "`" + String.raw`\r\n]+((?:"|'))`,
+9 -7
View File
@@ -139,6 +139,7 @@ import {
logActivity,
publishActivity,
projectService,
questionResponseDeliveryService,
routineService,
workProductService,
} from "../services/index.js";
@@ -2878,6 +2879,7 @@ export function issueRoutes(
const heartbeat = heartbeatService(db, {
pluginWorkerManager: opts.pluginWorkerManager,
});
const questionResponseDeliveries = questionResponseDeliveryService(db, { heartbeat });
const enqueueStalledReviewDecisionWakeup = opts.stalledReviewDecisionEnqueueWakeup ?? heartbeat.wakeup;
const enqueueRecoveryActionWakeup = opts.recoveryActionEnqueueWakeup ?? heartbeat.wakeup;
const feedback = feedbackService(db);
@@ -12226,13 +12228,13 @@ export function issueRoutes(
},
});
await queueResolvedInteractionContinuationWakeup({
db,
heartbeat,
issue,
interaction,
actor,
source: "issue.interaction.respond",
await questionResponseDeliveries.deliver(interaction.id).catch((err) => {
logger.warn({
err,
companyId: issue.companyId,
issueId: issue.id,
interactionId: interaction.id,
}, "synchronous question response delivery failed; durable outbox will retry");
});
res.json(interaction);
+1
View File
@@ -42,6 +42,7 @@ export {
type IssueFilters,
} from "./issues.js";
export { issueThreadInteractionService } from "./issue-thread-interactions.js";
export { questionResponseDeliveryService } from "./question-response-delivery.js";
export {
assertIssueReviewVerdictActorAllowed,
type IssueReviewVerdictActor,
@@ -9,6 +9,7 @@ import {
heartbeatRuns,
issueComments,
issueDocuments,
issueQuestionResponseDeliveries,
issueThreadInteractions,
issues,
toolActionRequests,
@@ -71,6 +72,7 @@ import {
isIssueReviewVerdictInteraction,
} from "./issue-review-policy.js";
import { issueService, runWorkspaceIsFinalized } from "./issues.js";
import { questionResponseDeliveryValues } from "./question-response-delivery.js";
import {
assertIssueThreadInteractionResolverAudience,
canonicalizeStoredResolverPolicy,
@@ -3377,30 +3379,36 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti
answers: input.answers,
});
const [updated] = await db
.update(issueThreadInteractions)
.set({
status: "answered",
result: {
version: 1,
answers: normalizedAnswers,
summaryMarkdown: input.summaryMarkdown ?? null,
},
resolvedByAgentId: actor.agentId ?? null,
resolvedByRunId: actor.runId ?? null,
resolvedByUserId: actor.userId ?? null,
resolvedAt: new Date(),
updatedAt: new Date(),
})
.where(and(
eq(issueThreadInteractions.id, interactionId),
eq(issueThreadInteractions.status, "pending"),
))
.returning();
const updated = await db.transaction(async (tx) => {
const resolvedAt = new Date();
const [row] = await tx
.update(issueThreadInteractions)
.set({
status: "answered",
result: {
version: 1,
answers: normalizedAnswers,
summaryMarkdown: input.summaryMarkdown ?? null,
},
resolvedByAgentId: actor.agentId ?? null,
resolvedByRunId: actor.runId ?? null,
resolvedByUserId: actor.userId ?? null,
resolvedAt,
updatedAt: resolvedAt,
})
.where(and(
eq(issueThreadInteractions.id, interactionId),
eq(issueThreadInteractions.status, "pending"),
))
.returning();
if (!updated) {
throw interactionAlreadyResolvedError();
}
if (!row) throw interactionAlreadyResolvedError();
const answered = hydrateInteraction(row) as AskUserQuestionsInteraction;
await tx.insert(issueQuestionResponseDeliveries).values(
questionResponseDeliveryValues(answered),
);
return row;
});
await touchIssue(db, issue.id);
const answered = hydrateInteraction(updated);
@@ -16,6 +16,7 @@ import { issueThreadInteractionService } from "../issue-thread-interactions.js";
import { commitNativeStatusDecision } from "./status-decision-committer.js";
import { nativeSha256 } from "./canonical.js";
import { recordNativeAttentionAssessment } from "./work-assessments.js";
import { formatDurableQuestionResponseSummary } from "../question-response-delivery.js";
import {
NATIVE_STATUS_ARBITER_POLICY_VERSION,
type NativeAuthoritativeIssueStatus,
@@ -319,12 +320,16 @@ export async function materializeNativeInteractionResponses(input: {
if (resolution.reasonCode !== "attention_resolved_from_context") {
throw new NativeInteractionBridgeError("native_attention_resolution_invalid", "Answered interaction has no continuation policy");
}
const result = structuredClone(interaction.result) as unknown as Record<string, unknown>;
const summaryMarkdown = formatDurableQuestionResponseSummary(interaction);
responses.push({
interactionId: interaction.id,
kind: interaction.kind,
response: {
status: interaction.status,
result: structuredClone(interaction.result) as unknown as Record<string, unknown>,
result: summaryMarkdown
? { ...result, summaryMarkdown }
: result,
},
});
continue;
@@ -27,6 +27,7 @@ vi.mock("@paperclipai/paperclip-runner", () => ({
CAPABILITY_SEMANTIC_TOOL_CATALOG: [],
createNativeSessionBackend: state.createBackend,
executeNativeSession: state.execute,
parsePaperclipQuestionSet: (value: unknown) => value,
}));
import {
@@ -42,10 +43,113 @@ import {
nativeSessionRecoveryProjection,
nativeGovernedWaitResult,
providerPlanMarkdown,
runtimeInputLifecycleMetric,
runtimeQuestionFallbackFromEvent,
semanticProviderPlanMarkdown,
steerNativeSession,
} from "./native-session-executor.js";
describe("runtime question fallback", () => {
const questionSet = {
schema: "paperclip.question_set.v1" as const,
title: "Configure deployment",
description: "These answers are required before work can continue.",
submitLabel: "Continue",
questions: [
{
id: "region",
prompt: "Which region?",
required: true,
answerMode: "single_select" as const,
options: [{ id: "us", label: "US" }, { id: "eu", label: "Europe" }],
},
{
id: "replicas",
prompt: "How many replicas?",
required: true,
answerMode: "text" as const,
textValidation: { inputType: "integer" as const, minimum: 1 },
},
],
};
it("materializes one idempotent durable interaction only after provider loss", () => {
const fallback = runtimeQuestionFallbackFromEvent({
eventType: "runtime_request.expired",
runId: "00000000-0000-4000-8000-000000000001",
payload: {
requestId: "elicitation-1",
reason: "provider_process_lost",
replayAllowed: false,
request: {
schema: "paperclip.runtime_request.v2",
requestKind: "runtime",
requestId: "elicitation-1",
type: "input",
status: "pending",
prompt: "Configure deployment",
input: questionSet,
},
},
});
expect(fallback).toMatchObject({
kind: "ask_user_questions",
idempotencyKey: "runtime-input-fallback:v1:00000000-0000-4000-8000-000000000001:elicitation-1",
sourceRunId: "00000000-0000-4000-8000-000000000001",
continuationPolicy: "wake_assignee",
payload: {
questionSet,
supersedeOnUserComment: false,
questions: [
{ id: "region", selectionMode: "single", options: [{ id: "us", label: "US" }, { id: "eu", label: "Europe" }] },
{ id: "replicas", selectionMode: "single", options: [{ id: "__paperclip_text__", freeText: true }] },
],
},
});
});
it.each([
["runtime_request.resolved", "provider_process_lost", false],
["runtime_request.cancelled", "provider_process_lost", false],
["runtime_request.expired", "explicit_cancellation", false],
["runtime_request.expired", "provider_process_lost", true],
])("does not fall back for %s / %s / replay=%s", (eventType, reason, replayAllowed) => {
expect(runtimeQuestionFallbackFromEvent({
eventType: eventType as never,
runId: "00000000-0000-4000-8000-000000000001",
payload: {
reason,
replayAllowed,
request: {
schema: "paperclip.runtime_request.v2",
requestKind: "runtime",
requestId: "elicitation-1",
type: "input",
input: questionSet,
},
},
})).toBeNull();
});
it("emits content-free lifecycle metric dimensions", () => {
expect(runtimeInputLifecycleMetric({
eventType: "runtime_request.created",
payload: {
request: {
type: "input",
requestId: "input-1",
origin: { adapter: "codex-app-server" },
input: questionSet,
},
},
})).toEqual({
outcome: "normalized",
adapter: "codex-app-server",
requestId: "input-1",
});
});
});
describe("native provider bootstrap environment", () => {
it("inherits the host executable and credential-home context", () => {
expect(buildNativeProviderEnvironment({}, {
@@ -9,13 +9,16 @@ import type {
NativeExecutionInput,
NativeSession,
NativeSessionBackend,
PaperclipQuestionSet,
PersistedNativeSession,
PrpEvent,
PrpStructuredRunResult,
} from "../../vendor/paperclip-runner/index.js";
import {
createNativeSessionBackend,
createRunnerdCodexTransport,
executeNativeSession,
parsePaperclipQuestionSet,
} from "../../vendor/paperclip-runner/index.js";
import type { Db } from "@paperclipai/db";
import { and, desc, eq, inArray, or, sql } from "drizzle-orm";
@@ -117,6 +120,128 @@ type PlanSynchronization = {
confirmationId: string | null;
};
type RuntimeQuestionFallback = {
kind: "ask_user_questions";
idempotencyKey: string;
sourceRunId: string;
title: string | null;
summary: string | null;
continuationPolicy: "wake_assignee";
payload: {
version: 1;
title?: string;
submitLabel?: string;
supersedeOnUserComment: false;
questionSet: PaperclipQuestionSet;
questions: Array<{
id: string;
prompt: string;
helpText?: string;
selectionMode: "single" | "multi";
required: boolean;
options: Array<{ id: string; label: string; description?: string; freeText?: boolean }>;
}>;
};
};
/** Translate only provider-loss expirations; cancellations and resolved inputs never fall back. */
export function runtimeQuestionFallbackFromEvent(
event: Pick<PrpEvent, "eventType" | "payload" | "runId">,
): RuntimeQuestionFallback | null {
if (event.eventType !== "runtime_request.expired") return null;
const payload = record(event.payload);
if (payload.reason !== "provider_process_lost" || payload.replayAllowed !== false) return null;
const request = record(payload.request);
if (
request.schema !== "paperclip.runtime_request.v2"
|| request.requestKind !== "runtime"
|| request.type !== "input"
|| typeof request.requestId !== "string"
) return null;
let questionSet: PaperclipQuestionSet;
try {
questionSet = parsePaperclipQuestionSet(request.input);
} catch {
return null;
}
const questions = questionSet.questions.map((question) => ({
id: question.id,
prompt: question.prompt,
...(question.helpText ? { helpText: question.helpText } : {}),
selectionMode: question.answerMode === "multi_select" ? "multi" as const : "single" as const,
required: question.required,
options: question.answerMode === "text"
? [{
id: "__paperclip_text__",
label: question.textValidation?.inputType === "integer"
? "Enter an integer"
: question.textValidation?.inputType === "number"
? "Enter a number"
: "Enter your answer",
freeText: true,
}]
: (question.options ?? []).map((option) => ({
id: option.id,
label: option.label,
...(option.description ? { description: option.description } : {}),
})),
}));
return {
kind: "ask_user_questions",
idempotencyKey: `runtime-input-fallback:v1:${event.runId}:${request.requestId}`,
sourceRunId: event.runId,
title: questionSet.title?.slice(0, 240) ?? null,
summary: questionSet.description?.slice(0, 1000) ?? null,
continuationPolicy: "wake_assignee",
payload: {
version: 1,
...(questionSet.title ? { title: questionSet.title.slice(0, 240) } : {}),
...(questionSet.submitLabel ? { submitLabel: questionSet.submitLabel.slice(0, 120) } : {}),
supersedeOnUserComment: false,
questionSet,
questions,
},
};
}
export function runtimeInputLifecycleMetric(
event: Pick<PrpEvent, "eventType" | "payload">,
): { outcome: "normalized" | "rejected" | "resolved" | "expired" | "cancelled"; adapter: string; requestId: string | null } | null {
const payload = record(event.payload);
const request = record(payload.request);
if (event.eventType === "runtime_request.created" && request.type === "input") {
const origin = record(request.origin);
return {
outcome: "normalized",
adapter: typeof origin.adapter === "string" ? origin.adapter : "unknown",
requestId: typeof request.requestId === "string" ? request.requestId : null,
};
}
if (event.eventType === "harness.diagnostic" && payload.code === "runtime_input_rejected") {
return {
outcome: "rejected",
adapter: typeof payload.adapter === "string" ? payload.adapter : "unknown",
requestId: null,
};
}
const terminalOutcome = event.eventType === "runtime_request.resolved" ? "resolved"
: event.eventType === "runtime_request.expired" ? "expired"
: event.eventType === "runtime_request.cancelled" ? "cancelled"
: null;
const requestType = payload.requestType ?? request.type;
if (!terminalOutcome || requestType !== "input") return null;
const origin = record(request.origin);
return {
outcome: terminalOutcome,
adapter: typeof payload.adapter === "string"
? payload.adapter
: typeof origin.adapter === "string" ? origin.adapter : "unknown",
requestId: typeof payload.requestId === "string"
? payload.requestId
: typeof request.requestId === "string" ? request.requestId : null,
};
}
export function providerPlanMarkdown(payload: Record<string, unknown>): string {
const completedMarkdown = typeof payload.markdown === "string" ? payload.markdown.trim() : "";
if (completedMarkdown) return completedMarkdown.slice(0, 256_000);
@@ -1082,6 +1207,38 @@ export async function executePaperclipNativeSession(input: {
}, {
onCommittedEvent: async (event) => {
if (input.onLog) await input.onLog("stdout", `${JSON.stringify({ type: "paperclip.prp.event", event })}\n`);
const inputMetric = runtimeInputLifecycleMetric(event);
if (inputMetric && input.onLog) {
await input.onLog("stdout", `${JSON.stringify({
type: "paperclip.runtime_input.metric",
...inputMetric,
})}\n`);
}
const questionFallback = runtimeQuestionFallbackFromEvent(event);
if (questionFallback) {
const interaction = await issueThreadInteractionService(input.db).create(
{
id: input.execution.binding.issueId,
companyId: input.execution.binding.companyId,
},
questionFallback as never,
{
agentId: input.execution.binding.agentId,
runId: input.execution.binding.runId,
systemId: "native-runtime-question-fallback",
},
);
if (input.onLog) {
const origin = record(record(record(event.payload).request).origin);
await input.onLog("stdout", `${JSON.stringify({
type: "paperclip.runtime_input.metric",
outcome: "fallback_materialized",
requestId: record(event.payload).requestId,
interactionId: interaction.id,
adapter: typeof origin.adapter === "string" ? origin.adapter : "unknown",
})}\n`);
}
}
await recordPlanSynchronization(event as {
sourceEventId: string;
turnId?: string;
@@ -0,0 +1,532 @@
import { and, asc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
agents,
heartbeatRuns,
issueQuestionResponseDeliveries,
issues,
issueThreadInteractions,
} from "@paperclipai/db";
import type {
AskUserQuestionsInteraction,
PaperclipQuestionSetPayload,
} from "@paperclipai/shared";
import type {
PaperclipQuestionResponse,
} from "../vendor/paperclip-runner/index.js";
import { getTelemetryClient } from "../telemetry.js";
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 {
NativeSessionSteeringError,
steerNativeSession,
} from "./native-runtime/native-session-executor.js";
const DELIVERY_CLAIM_STALE_MS = 30_000;
const MAX_DELIVERY_ATTEMPTS = 5;
const DELIVERY_CORRELATION_PREFIX = "question-response:";
type QuestionInteractionRow = typeof issueThreadInteractions.$inferSelect;
type DeliveryRow = typeof issueQuestionResponseDeliveries.$inferSelect;
type Heartbeat = Pick<ReturnType<typeof heartbeatService>, "wakeup">;
export interface QuestionResponseDeliveryEnvelope {
schema: "paperclip.question_response_delivery.v1";
interactionId: string;
sourceRunId: string | null;
questionSet: PaperclipQuestionSetPayload;
response: PaperclipQuestionResponse;
}
export interface QuestionResponseDeliveryOutcome {
deliveryId: string;
status: DeliveryRow["status"];
mode: DeliveryRow["deliveryMode"];
targetRunId: string | null;
targetTurnId: string | null;
duplicate: boolean;
}
export interface QuestionResponseDeliveryServiceOptions {
heartbeat: Heartbeat;
steer?: typeof steerNativeSession;
now?: () => Date;
}
function record(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
function compactLine(value: unknown): string | null {
if (typeof value !== "string") return null;
const normalized = value.replace(/\s+/g, " ").trim();
return normalized.length > 0 ? normalized : null;
}
function canonicalQuestionSet(interaction: Pick<AskUserQuestionsInteraction, "title" | "payload">): PaperclipQuestionSetPayload {
if (interaction.payload.questionSet) return structuredClone(interaction.payload.questionSet);
return {
schema: "paperclip.question_set.v1",
...(interaction.title ? { title: interaction.title } : {}),
...(interaction.payload.submitLabel ? { submitLabel: interaction.payload.submitLabel } : {}),
questions: interaction.payload.questions.map((question) => {
const customOption = question.options.find((option) => option.freeText === true);
return {
id: question.id,
prompt: question.prompt,
...(question.helpText ? { helpText: question.helpText } : {}),
required: question.required === true,
answerMode: question.selectionMode === "multi" ? "multi_select" as const : "single_select" as const,
options: question.options
.filter((option) => option.freeText !== true)
.map((option) => ({
id: option.id,
label: option.label,
...(option.description ? { description: option.description } : {}),
})),
...(customOption
? {
customAnswer: {
enabled: true as const,
label: customOption.label,
...(customOption.description ? { placeholder: customOption.description } : {}),
},
}
: {}),
};
}),
};
}
export function buildQuestionResponseDeliveryEnvelope(
interaction: AskUserQuestionsInteraction,
): QuestionResponseDeliveryEnvelope {
if (interaction.status !== "answered" || !interaction.result || interaction.result.cancelled === true) {
throw new Error("question_response_interaction_not_answered");
}
const questionSet = canonicalQuestionSet(interaction);
const questionById = new Map(questionSet.questions.map((question) => [question.id, question]));
const response: PaperclipQuestionResponse = {
schema: "paperclip.question_response.v1",
answers: Object.fromEntries(interaction.result.answers.map((answer) => {
const question = questionById.get(answer.questionId);
return [answer.questionId, question?.answerMode === "text"
? { ...(answer.otherText ? { text: answer.otherText } : {}) }
: {
selectedOptionIds: answer.optionIds,
...(answer.otherText ? { customText: answer.otherText } : {}),
}];
})),
};
return {
schema: "paperclip.question_response_delivery.v1",
interactionId: interaction.id,
sourceRunId: interaction.sourceRunId ?? null,
questionSet,
response,
};
}
function questionAnswerLines(envelope: QuestionResponseDeliveryEnvelope): string[] {
const lines: string[] = [];
for (const question of envelope.questionSet.questions) {
const answer = envelope.response.answers[question.id];
if (!answer) continue;
const optionLabelById = new Map((question.options ?? []).map((option) => [option.id, option.label]));
const values = (answer.selectedOptionIds ?? []).map((optionId) => optionLabelById.get(optionId) ?? optionId);
const text = compactLine(answer.text);
const customText = compactLine(answer.customText);
if (text) values.push(text);
if (customText) values.push(customText);
const header = compactLine(question.header);
const prompt = compactLine(question.prompt);
const label = header && prompt && header !== prompt
? `${header} — ${prompt}`
: header ?? prompt ?? question.id;
lines.push(`- ${label}: ${values.join(", ") || "No answer"}`);
}
return lines;
}
export function formatQuestionResponseSummary(envelope: QuestionResponseDeliveryEnvelope): string {
const lines = questionAnswerLines(envelope);
return lines.length > 0
? ["Resolved questions and answers:", ...lines].join("\n")
: "Resolved questions and answers.";
}
export function formatDurableQuestionResponseSummary(interaction: AskUserQuestionsInteraction): string {
const existing = compactLine(interaction.result?.summaryMarkdown);
return existing ?? formatQuestionResponseSummary(buildQuestionResponseDeliveryEnvelope(interaction));
}
export function formatQuestionResponseSteeringMessage(envelope: QuestionResponseDeliveryEnvelope): string {
const lines = questionAnswerLines(envelope);
return lines.length > 0
? ["Answered questions", "", ...lines].join("\n")
: "Answered questions";
}
function hydrateQuestionInteraction(row: QuestionInteractionRow): AskUserQuestionsInteraction {
return {
...row,
kind: "ask_user_questions",
status: row.status as AskUserQuestionsInteraction["status"],
continuationPolicy: row.continuationPolicy as AskUserQuestionsInteraction["continuationPolicy"],
resolverPolicy: row.effectiveResolverPolicy,
requestedResolverPolicy: row.requestedResolverPolicy,
effectiveResolverPolicy: row.effectiveResolverPolicy,
resolverPolicyProvenance: row.resolverPolicyProvenance,
effectiveResolverPolicySource: row.effectiveResolverPolicySource,
legacyResolverPolicyAliases: { requested: null, effective: null },
payload: row.payload as AskUserQuestionsInteraction["payload"],
result: row.result as AskUserQuestionsInteraction["result"],
};
}
export function questionResponseDeliveryValues(interaction: AskUserQuestionsInteraction) {
const envelope = buildQuestionResponseDeliveryEnvelope(interaction);
return {
companyId: interaction.companyId,
issueId: interaction.issueId,
interactionId: interaction.id,
sourceRunId: interaction.sourceRunId ?? null,
correlationId: `${DELIVERY_CORRELATION_PREFIX}${interaction.id}`,
payloadSha256: nativeSha256(envelope),
};
}
function issueIdFromRun(run: Pick<typeof heartbeatRuns.$inferSelect, "contextSnapshot">) {
const context = record(run.contextSnapshot);
return compactLine(context.issueId) ?? compactLine(context.taskId);
}
function actorForInteraction(interaction: QuestionInteractionRow) {
if (interaction.resolvedByUserId) {
return { actorType: "user" as const, actorId: interaction.resolvedByUserId };
}
if (interaction.resolvedByAgentId) {
return { actorType: "agent" as const, actorId: interaction.resolvedByAgentId };
}
return { actorType: "system" as const, actorId: "question-response-outbox" };
}
export function questionResponseDeliveryService(
db: Db,
options: QuestionResponseDeliveryServiceOptions,
) {
const steer = options.steer ?? steerNativeSession;
const now = options.now ?? (() => new Date());
async function claim(interactionId: string): Promise<DeliveryRow | null> {
const claimAt = now();
return db.transaction(async (tx) => {
const current = await tx.select()
.from(issueQuestionResponseDeliveries)
.where(eq(issueQuestionResponseDeliveries.interactionId, interactionId))
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null);
if (!current || ["delivered", "fallback_queued", "failed"].includes(current.status)) return null;
if (
current.status === "delivering"
&& current.lastAttemptAt
&& current.lastAttemptAt.getTime() > claimAt.getTime() - DELIVERY_CLAIM_STALE_MS
) return null;
return tx.update(issueQuestionResponseDeliveries).set({
status: "delivering",
attemptCount: sql`${issueQuestionResponseDeliveries.attemptCount} + 1`,
lastAttemptAt: claimAt,
updatedAt: claimAt,
}).where(eq(issueQuestionResponseDeliveries.id, current.id))
.returning()
.then((rows) => rows[0] ?? null);
});
}
async function terminalOutcome(interactionId: string): Promise<QuestionResponseDeliveryOutcome | null> {
const row = await db.select().from(issueQuestionResponseDeliveries)
.where(eq(issueQuestionResponseDeliveries.interactionId, interactionId))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!row || !["delivered", "fallback_queued", "failed"].includes(row.status)) return null;
return {
deliveryId: row.id,
status: row.status,
mode: row.deliveryMode,
targetRunId: row.targetRunId,
targetTurnId: row.targetTurnId,
duplicate: true,
};
}
async function recordTerminal(input: {
delivery: DeliveryRow;
interaction: QuestionInteractionRow;
status: "delivered" | "fallback_queued" | "failed";
mode: "steered" | "coalesced" | "wake_fallback" | null;
targetRunId: string | null;
targetTurnId?: string | null;
adapter: string;
errorCode?: string | null;
}): Promise<QuestionResponseDeliveryOutcome> {
const at = now();
const updated = await db.update(issueQuestionResponseDeliveries).set({
status: input.status,
deliveryMode: input.mode,
targetRunId: input.targetRunId,
targetTurnId: input.targetTurnId ?? null,
acknowledgedAt: input.status === "failed" ? null : at,
lastErrorCode: input.errorCode ?? null,
updatedAt: at,
}).where(and(
eq(issueQuestionResponseDeliveries.id, input.delivery.id),
eq(issueQuestionResponseDeliveries.status, "delivering"),
)).returning().then((rows) => rows[0] ?? null);
const result = updated ?? await db.select().from(issueQuestionResponseDeliveries)
.where(eq(issueQuestionResponseDeliveries.id, input.delivery.id))
.limit(1)
.then((rows) => rows[0] ?? input.delivery);
if (updated) {
await logActivity(db, {
companyId: input.interaction.companyId,
actorType: "system",
actorId: "question-response-delivery",
agentId: input.interaction.resolvedByAgentId,
runId: input.targetRunId,
action: input.status === "failed"
? "issue.question_response_delivery_failed"
: "issue.question_response_delivered",
entityType: "issue",
entityId: input.interaction.issueId,
details: {
deliveryId: result.id,
interactionId: input.interaction.id,
sourceRunId: input.interaction.sourceRunId,
targetRunId: input.targetRunId,
targetTurnId: input.targetTurnId ?? null,
correlationId: result.correlationId,
payloadSha256: result.payloadSha256,
deliveryStatus: input.status,
deliveryMode: input.mode,
adapter: input.adapter,
errorCode: input.errorCode ?? null,
},
});
getTelemetryClient()?.trackDynamic("question_response.delivery", {
adapter: input.adapter,
outcome: input.mode ?? "failed",
});
}
return {
deliveryId: result.id,
status: result.status,
mode: result.deliveryMode,
targetRunId: result.targetRunId,
targetTurnId: result.targetTurnId,
duplicate: !updated,
};
}
async function releaseForRetry(delivery: DeliveryRow, errorCode: string) {
const at = now();
const exhausted = delivery.attemptCount >= MAX_DELIVERY_ATTEMPTS;
await db.update(issueQuestionResponseDeliveries).set({
status: exhausted ? "failed" : "pending",
lastErrorCode: errorCode,
updatedAt: at,
}).where(and(
eq(issueQuestionResponseDeliveries.id, delivery.id),
eq(issueQuestionResponseDeliveries.status, "delivering"),
));
return exhausted;
}
async function deliver(interactionId: string): Promise<QuestionResponseDeliveryOutcome | null> {
const claimed = await claim(interactionId);
if (!claimed) return terminalOutcome(interactionId);
const interaction = await db.select().from(issueThreadInteractions)
.where(and(
eq(issueThreadInteractions.id, interactionId),
eq(issueThreadInteractions.companyId, claimed.companyId),
eq(issueThreadInteractions.issueId, claimed.issueId),
))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!interaction || interaction.kind !== "ask_user_questions" || interaction.status !== "answered") {
return recordTerminal({
delivery: claimed,
interaction: interaction ?? ({
id: interactionId,
companyId: claimed.companyId,
issueId: claimed.issueId,
sourceRunId: claimed.sourceRunId,
resolvedByAgentId: null,
} as QuestionInteractionRow),
status: "failed",
mode: null,
targetRunId: null,
adapter: "unknown",
errorCode: "question_response_interaction_invalid",
});
}
const [issue, agent] = await Promise.all([
db.select().from(issues).where(and(
eq(issues.id, interaction.issueId),
eq(issues.companyId, interaction.companyId),
)).limit(1).then((rows) => rows[0] ?? null),
interaction.createdByAgentId
? db.select({ adapterType: agents.adapterType }).from(agents)
.where(and(eq(agents.id, interaction.createdByAgentId), eq(agents.companyId, interaction.companyId)))
.limit(1).then((rows) => rows[0] ?? null)
: Promise.resolve(null),
]);
const adapter = agent?.adapterType ?? "unknown";
if (!issue || !issue.assigneeAgentId || issue.status === "done" || issue.status === "cancelled") {
return recordTerminal({
delivery: claimed,
interaction,
status: "failed",
mode: null,
targetRunId: null,
adapter,
errorCode: !issue ? "question_response_issue_missing" : "question_response_target_unavailable",
});
}
const liveRuns = await db.select().from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, interaction.companyId),
eq(heartbeatRuns.agentId, issue.assigneeAgentId),
inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
)).orderBy(asc(heartbeatRuns.createdAt));
const issueRuns = liveRuns.filter((run) => issueIdFromRun(run) === interaction.issueId);
const successorRunning = issueRuns.find((run) =>
run.status === "running" && run.id !== interaction.sourceRunId,
) ?? null;
const queuedSuccessor = issueRuns.find((run) =>
(run.status === "queued" || run.status === "scheduled_retry") && run.id !== interaction.sourceRunId,
) ?? null;
const envelope = buildQuestionResponseDeliveryEnvelope(hydrateQuestionInteraction(interaction));
let steeringErrorCode: string | null = null;
if (successorRunning?.runtimeMode === "native") {
try {
const acknowledgement = await steer({
runId: successorRunning.id,
message: formatQuestionResponseSteeringMessage(envelope),
correlationId: claimed.correlationId,
});
return recordTerminal({
delivery: claimed,
interaction,
status: "delivered",
mode: "steered",
targetRunId: successorRunning.id,
targetTurnId: acknowledgement.turnId,
adapter: successorRunning.driverKind ?? adapter,
});
} catch (error) {
steeringErrorCode = error instanceof NativeSessionSteeringError
? error.code
: "steering_rejected";
}
} else if (successorRunning) {
steeringErrorCode = "steering_unsupported";
}
const actor = actorForInteraction(interaction);
try {
const wakeRun = await options.heartbeat.wakeup(issue.assigneeAgentId, {
source: "automation",
triggerDetail: "system",
reason: "issue_commented",
payload: {
issueId: issue.id,
interactionId: interaction.id,
interactionKind: interaction.kind,
interactionStatus: interaction.status,
sourceCommentId: interaction.sourceCommentId,
sourceRunId: interaction.sourceRunId,
mutation: "interaction",
},
idempotencyKey: `interaction:${interaction.id}:${interaction.status}`,
requestedByActorType: actor.actorType,
requestedByActorId: actor.actorId,
contextSnapshot: {
issueId: issue.id,
taskId: issue.id,
interactionId: interaction.id,
interactionKind: interaction.kind,
interactionStatus: interaction.status,
sourceCommentId: interaction.sourceCommentId,
sourceRunId: interaction.sourceRunId,
wakeReason: "issue_commented",
source: "issue.interaction.respond",
},
});
const coalesced = Boolean(queuedSuccessor && wakeRun?.id === queuedSuccessor.id);
return recordTerminal({
delivery: claimed,
interaction,
status: coalesced ? "delivered" : "fallback_queued",
mode: coalesced ? "coalesced" : "wake_fallback",
targetRunId: wakeRun?.id ?? queuedSuccessor?.id ?? null,
adapter: wakeRun?.driverKind ?? adapter,
errorCode: steeringErrorCode,
});
} catch (error) {
const errorCode = error instanceof Error && compactLine(error.message)
? compactLine(error.message)!.slice(0, 160)
: "question_response_wake_failed";
const exhausted = await releaseForRetry(claimed, errorCode);
logger.warn({
err: error,
deliveryId: claimed.id,
interactionId,
attemptCount: claimed.attemptCount,
exhausted,
}, "question response delivery will retry after wake failure");
if (!exhausted) return null;
return terminalOutcome(interactionId);
}
}
async function sweepPending(limit = 50) {
const sweepAt = now();
const staleAt = new Date(sweepAt.getTime() - DELIVERY_CLAIM_STALE_MS);
await db.update(issueQuestionResponseDeliveries).set({
status: "pending",
updatedAt: sweepAt,
}).where(and(
eq(issueQuestionResponseDeliveries.status, "delivering"),
or(
isNull(issueQuestionResponseDeliveries.lastAttemptAt),
lte(issueQuestionResponseDeliveries.lastAttemptAt, staleAt),
),
));
const ids = await db.select({ interactionId: issueQuestionResponseDeliveries.interactionId })
.from(issueQuestionResponseDeliveries)
.where(eq(issueQuestionResponseDeliveries.status, "pending"))
.orderBy(asc(issueQuestionResponseDeliveries.createdAt))
.limit(limit)
.then((rows) => rows.map((row) => row.interactionId));
const counts = { scanned: ids.length, steered: 0, coalesced: 0, wakeFallback: 0, failed: 0 };
for (const id of ids) {
const outcome = await deliver(id);
if (outcome?.mode === "steered") counts.steered += 1;
else if (outcome?.mode === "coalesced") counts.coalesced += 1;
else if (outcome?.mode === "wake_fallback") counts.wakeFallback += 1;
else if (outcome?.status === "failed") counts.failed += 1;
}
return counts;
}
return { deliver, sweepPending };
}