fix(interactions): deliver question answers durably (#12307)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agents can pause a task and ask the user structured questions. > - The answer is durable in the issue interaction, but delivery to the next run is not durable. > - A process restart can therefore leave an answered interaction without a continuation attempt. > - Native runners also need a provider-neutral question contract before the task page can consume native events safely. > - This pull request adds a content-free delivery outbox and an optional native steering seam. > - Direct adapters keep their existing heartbeat continuation path. > - The benefit is reliable answer delivery without changing runtime selection or task-page behavior. ## Linked Issues or Issue Description Refs #12202. This pull request replaces the question-delivery foundation from that stale task-thread pull request. The task-thread projection will follow in a smaller pull request. **What happened?** Question answers were stored in the issue interaction. The server then made one in-memory continuation wake. A server stop between those operations could leave the answer stored but not delivered. The combined native task-thread pull request also made this behavior hard to review separately from UI changes. **Expected behavior** The answer and its delivery receipt must commit in one transaction. The server must retry pending receipts after a restart. Existing direct adapters must keep the current wake path. A native runtime may use the optional steering seam, but this pull request does not enable native steering in production. **Steps to reproduce** 1. Create an `ask_user_questions` interaction. 2. Answer the interaction. 3. Stop the server before the continuation wake completes. 4. Start the server again. 5. On current master, no durable record tells the server to retry the answer delivery. **Paperclip version or commit** Current `master` at `4d82f5eae`. ## What Changed - Add the `issue_question_response_deliveries` table and migration. - Store only routing state, a correlation ID, and a payload digest in the delivery row. The answer remains in the existing interaction result. - Commit an answered interaction and its pending delivery row in one transaction. - Add bounded claims, retry recovery, cumulative terminal state, and content-free activity records. - Keep every built-in direct adapter and external adapter on the existing heartbeat wake path. - Add an optional native steering seam. No production caller supplies that seam in this pull request. - Retain the provider-neutral `paperclip.question_set.v1` presentation on recovered interactions. - Run delivery immediately after an answer and sweep pending rows at startup and on the existing server interval. - Add focused database, service, route, startup, adapter-matrix, digest, and duplicate-delivery tests. ## Compatibility Boundary - This pull request does not change adapter selection. - This pull request does not start runnerd. - This pull request does not create native run records. - Direct adapters never call the native steering seam. - The existing interaction result stays authoritative for answer content. - The migration is additive and does not rewrite existing rows. - This pull request has no UI, dependency, workflow, package-manager, or lockfile changes. - The diff has 19 files. ## Verification - `pnpm exec vitest run server/src/__tests__/question-response-delivery.test.ts server/src/services/issue-thread-interactions.test.ts server/src/__tests__/issue-thread-interaction-routes.test.ts server/src/__tests__/server-startup-feedback-export.test.ts` — 4 files and 120 tests passed. - `pnpm -r typecheck` — passed for all applicable workspaces. This includes Cargo format and check, protocol drift checks, and migration safety. - `pnpm build` — passed. This includes the Rust release binary, server build, and UI production build. - `git diff --check` — passed. - Secret patterns were not present in the changed text files. - The repository token gates currently report violations from unchanged files on `master`. This pull request does not change those files. ## Risks The main risk is routing a direct-adapter answer into a native session. The service checks the persisted runtime mode, and the adapter matrix proves that all direct adapters use only the existing wake path. The new table is additive. It has foreign keys, unique correlation constraints, bounded attempts, and status checks. Activity records omit question and answer content. ## Model Used OpenAI Codex, GPT-5 family. The client does not expose the exact deployment ID or context window. Agentic reasoning, tool use, and code execution were enabled. ## 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 or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket ID or instance-derived details - [x] I have run the affected tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have documented the new contracts and compatibility boundary - [x] I have considered and documented compatibility and security risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
c9df0e251d
commit
67f9867bc6
|
|
@ -0,0 +1,59 @@
|
|||
CREATE TABLE IF NOT EXISTS "issue_question_response_deliveries" (
|
||||
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
|
||||
"company_id" uuid NOT NULL,
|
||||
"issue_id" uuid NOT NULL,
|
||||
"interaction_id" uuid NOT NULL,
|
||||
"source_run_id" uuid,
|
||||
"target_run_id" uuid,
|
||||
"target_turn_id" text,
|
||||
"correlation_id" text NOT NULL,
|
||||
"payload_sha256" text NOT NULL,
|
||||
"status" text DEFAULT 'pending' NOT NULL,
|
||||
"delivery_mode" text,
|
||||
"attempt_count" integer DEFAULT 0 NOT NULL,
|
||||
"error_count" integer DEFAULT 0 NOT NULL,
|
||||
"last_attempt_at" timestamp with time zone,
|
||||
"acknowledged_at" timestamp with time zone,
|
||||
"last_error_code" text,
|
||||
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
|
||||
"updated_at" timestamp with time zone DEFAULT now() NOT NULL,
|
||||
CONSTRAINT "issue_question_response_deliveries_status_check" CHECK ("issue_question_response_deliveries"."status" IN ('pending', 'delivering', 'delivered', 'fallback_queued', 'failed')),
|
||||
CONSTRAINT "issue_question_response_deliveries_mode_check" CHECK ("issue_question_response_deliveries"."delivery_mode" IS NULL OR "issue_question_response_deliveries"."delivery_mode" IN ('steered', 'coalesced', 'wake_fallback'))
|
||||
);
|
||||
--> statement-breakpoint
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE "issue_question_response_deliveries" ADD CONSTRAINT "issue_question_response_deliveries_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE cascade ON UPDATE no action;
|
||||
EXCEPTION
|
||||
WHEN duplicate_object THEN null;
|
||||
END $$;
|
||||
--> statement-breakpoint
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE "issue_question_response_deliveries" ADD CONSTRAINT "issue_question_response_deliveries_issue_id_issues_id_fk" FOREIGN KEY ("issue_id") REFERENCES "public"."issues"("id") ON DELETE cascade ON UPDATE no action;
|
||||
EXCEPTION
|
||||
WHEN duplicate_object THEN null;
|
||||
END $$;
|
||||
--> statement-breakpoint
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE "issue_question_response_deliveries" ADD CONSTRAINT "issue_question_response_deliveries_interaction_id_issue_thread_interactions_id_fk" FOREIGN KEY ("interaction_id") REFERENCES "public"."issue_thread_interactions"("id") ON DELETE cascade ON UPDATE no action;
|
||||
EXCEPTION
|
||||
WHEN duplicate_object THEN null;
|
||||
END $$;
|
||||
--> statement-breakpoint
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE "issue_question_response_deliveries" ADD CONSTRAINT "issue_question_response_deliveries_source_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("source_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action;
|
||||
EXCEPTION
|
||||
WHEN duplicate_object THEN null;
|
||||
END $$;
|
||||
--> statement-breakpoint
|
||||
DO $$ BEGIN
|
||||
ALTER TABLE "issue_question_response_deliveries" ADD CONSTRAINT "issue_question_response_deliveries_target_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("target_run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE set null ON UPDATE no action;
|
||||
EXCEPTION
|
||||
WHEN duplicate_object THEN null;
|
||||
END $$;
|
||||
--> statement-breakpoint
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS "issue_question_response_deliveries_interaction_uq" ON "issue_question_response_deliveries" USING btree ("interaction_id");--> statement-breakpoint
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS "issue_question_response_deliveries_correlation_uq" ON "issue_question_response_deliveries" USING btree ("correlation_id");--> statement-breakpoint
|
||||
CREATE INDEX IF NOT EXISTS "issue_question_response_deliveries_pending_idx" ON "issue_question_response_deliveries" USING btree ("status","created_at");--> statement-breakpoint
|
||||
CREATE INDEX IF NOT EXISTS "issue_question_response_deliveries_company_issue_idx" ON "issue_question_response_deliveries" USING btree ("company_id","issue_id","created_at");--> statement-breakpoint
|
||||
-- paperclip:migration-safety-ignore large-create-index-not-concurrently: This partial index covers a new question-response key prefix, so existing deployments have no matching rows; Drizzle applies migrations transactionally and cannot use CONCURRENTLY.
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS "agent_wakeup_requests_question_response_delivery_idempotency_uq" ON "agent_wakeup_requests" USING btree ("company_id","idempotency_key") WHERE "agent_wakeup_requests"."idempotency_key" LIKE 'question-response:%' AND "agent_wakeup_requests"."status" NOT IN ('skipped', 'failed', 'cancelled');
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -1583,6 +1583,13 @@
|
|||
"when": 1787668790923,
|
||||
"tag": "0227_modern_pandemic",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 228,
|
||||
"version": "7",
|
||||
"when": 1787837657366,
|
||||
"tag": "0228_nasty_grim_reaper",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -0,0 +1,78 @@
|
|||
import { readFile } from "node:fs/promises";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import postgres from "postgres";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./test-embedded-postgres.js";
|
||||
|
||||
const cleanups: Array<() => Promise<void>> = [];
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = support.supported ? describe : describe.skip;
|
||||
|
||||
afterEach(async () => {
|
||||
while (cleanups.length > 0) await cleanups.pop()?.();
|
||||
});
|
||||
|
||||
async function migrationStatements(): Promise<string[]> {
|
||||
const migrationSql = await readFile(
|
||||
fileURLToPath(
|
||||
new URL("./migrations/0228_nasty_grim_reaper.sql", import.meta.url),
|
||||
),
|
||||
"utf8",
|
||||
);
|
||||
return migrationSql
|
||||
.split("--> statement-breakpoint")
|
||||
.map((statement) => statement.trim())
|
||||
.filter((statement) => statement.length > 0);
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("question response delivery migration", () => {
|
||||
it("can be replayed against an already migrated database", async () => {
|
||||
const database = await startEmbeddedPostgresTestDatabase(
|
||||
"question-response-migration-",
|
||||
);
|
||||
cleanups.push(database.cleanup);
|
||||
const sql = postgres(database.connectionString, {
|
||||
max: 1,
|
||||
onnotice: () => {},
|
||||
});
|
||||
cleanups.push(async () => sql.end());
|
||||
|
||||
const statements = await migrationStatements();
|
||||
expect(statements.length).toBeGreaterThan(0);
|
||||
|
||||
for (const statement of statements) await sql.unsafe(statement);
|
||||
for (const statement of statements) await sql.unsafe(statement);
|
||||
|
||||
const [table] = await sql<{ table_name: string }[]>`
|
||||
SELECT table_name
|
||||
FROM information_schema.tables
|
||||
WHERE table_schema = 'public'
|
||||
AND table_name = 'issue_question_response_deliveries'
|
||||
`;
|
||||
expect(table?.table_name).toBe("issue_question_response_deliveries");
|
||||
|
||||
const indexes = await sql<{ indexname: string }[]>`
|
||||
SELECT indexname
|
||||
FROM pg_indexes
|
||||
WHERE indexname IN (
|
||||
'issue_question_response_deliveries_interaction_uq',
|
||||
'issue_question_response_deliveries_correlation_uq',
|
||||
'issue_question_response_deliveries_pending_idx',
|
||||
'issue_question_response_deliveries_company_issue_idx',
|
||||
'agent_wakeup_requests_question_response_delivery_idempotency_uq'
|
||||
)
|
||||
`;
|
||||
expect(indexes.map((row) => row.indexname).sort()).toEqual(
|
||||
[
|
||||
"agent_wakeup_requests_question_response_delivery_idempotency_uq",
|
||||
"issue_question_response_deliveries_company_issue_idx",
|
||||
"issue_question_response_deliveries_correlation_uq",
|
||||
"issue_question_response_deliveries_interaction_uq",
|
||||
"issue_question_response_deliveries_pending_idx",
|
||||
].sort(),
|
||||
);
|
||||
}, 240_000);
|
||||
});
|
||||
|
|
@ -43,6 +43,11 @@ export const agentWakeupRequests = pgTable(
|
|||
dispositionRepairIdempotencyUq: uniqueIndex("agent_wakeup_requests_disposition_repair_idempotency_uq")
|
||||
.on(table.companyId, table.idempotencyKey)
|
||||
.where(sql`${table.idempotencyKey} LIKE 'issue_disposition_repair:%' AND ${table.status} <> 'skipped'`),
|
||||
questionResponseDeliveryIdempotencyUq: uniqueIndex(
|
||||
"agent_wakeup_requests_question_response_delivery_idempotency_uq",
|
||||
)
|
||||
.on(table.companyId, table.idempotencyKey)
|
||||
.where(sql`${table.idempotencyKey} LIKE 'question-response:%' AND ${table.status} NOT IN ('skipped', 'failed', 'cancelled')`),
|
||||
companyPayloadIssueIdx: index("agent_wakeup_requests_company_payload_issue_idx").on(
|
||||
table.companyId,
|
||||
sql`(${table.payload} ->> 'issueId')`,
|
||||
|
|
|
|||
|
|
@ -74,6 +74,7 @@ export { issueApprovals } from "./issue_approvals.js";
|
|||
export { issueComments } from "./issue_comments.js";
|
||||
export { issueCreateIdempotencyKeys } from "./issue_create_idempotency_keys.js";
|
||||
export { issueThreadInteractions } from "./issue_thread_interactions.js";
|
||||
export { issueQuestionResponseDeliveries } from "./issue_question_response_deliveries.js";
|
||||
export {
|
||||
decisions,
|
||||
decisionBundles,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,65 @@
|
|||
import { sql } from "drizzle-orm";
|
||||
import {
|
||||
check,
|
||||
index,
|
||||
integer,
|
||||
pgTable,
|
||||
text,
|
||||
timestamp,
|
||||
uniqueIndex,
|
||||
uuid,
|
||||
} from "drizzle-orm/pg-core";
|
||||
import { companies } from "./companies.js";
|
||||
import { heartbeatRuns } from "./heartbeat_runs.js";
|
||||
import { issues } from "./issues.js";
|
||||
import { issueThreadInteractions } from "./issue_thread_interactions.js";
|
||||
|
||||
/**
|
||||
* Durable, content-free delivery state for an answered question interaction.
|
||||
* The answer remains authoritative in issue_thread_interactions.result; this
|
||||
* row stores only routing state and a digest of the canonical delivery
|
||||
* envelope so retries cannot duplicate or silently change the message.
|
||||
*/
|
||||
export const issueQuestionResponseDeliveries = pgTable(
|
||||
"issue_question_response_deliveries",
|
||||
{
|
||||
id: uuid("id").primaryKey().defaultRandom(),
|
||||
companyId: uuid("company_id").notNull().references(() => companies.id, { onDelete: "cascade" }),
|
||||
issueId: uuid("issue_id").notNull().references(() => issues.id, { onDelete: "cascade" }),
|
||||
interactionId: uuid("interaction_id").notNull().references(() => issueThreadInteractions.id, { onDelete: "cascade" }),
|
||||
sourceRunId: uuid("source_run_id").references(() => heartbeatRuns.id, { onDelete: "set null" }),
|
||||
targetRunId: uuid("target_run_id").references(() => heartbeatRuns.id, { onDelete: "set null" }),
|
||||
targetTurnId: text("target_turn_id"),
|
||||
correlationId: text("correlation_id").notNull(),
|
||||
payloadSha256: text("payload_sha256").notNull(),
|
||||
status: text("status").notNull().default("pending"),
|
||||
deliveryMode: text("delivery_mode"),
|
||||
/** Monotonic claim generation used to fence stale workers. */
|
||||
attemptCount: integer("attempt_count").notNull().default(0),
|
||||
/** Actual side-effect failures; scheduling suppression does not consume this budget. */
|
||||
errorCount: integer("error_count").notNull().default(0),
|
||||
lastAttemptAt: timestamp("last_attempt_at", { withTimezone: true }),
|
||||
acknowledgedAt: timestamp("acknowledged_at", { withTimezone: true }),
|
||||
lastErrorCode: text("last_error_code"),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(table) => ({
|
||||
interactionUq: uniqueIndex("issue_question_response_deliveries_interaction_uq").on(table.interactionId),
|
||||
correlationUq: uniqueIndex("issue_question_response_deliveries_correlation_uq").on(table.correlationId),
|
||||
pendingIdx: index("issue_question_response_deliveries_pending_idx").on(table.status, table.createdAt),
|
||||
companyIssueIdx: index("issue_question_response_deliveries_company_issue_idx").on(
|
||||
table.companyId,
|
||||
table.issueId,
|
||||
table.createdAt,
|
||||
),
|
||||
statusCheck: check(
|
||||
"issue_question_response_deliveries_status_check",
|
||||
sql`${table.status} IN ('pending', 'delivering', 'delivered', 'fallback_queued', 'failed')`,
|
||||
),
|
||||
modeCheck: check(
|
||||
"issue_question_response_deliveries_mode_check",
|
||||
sql`${table.deliveryMode} IS NULL OR ${table.deliveryMode} IN ('steered', 'coalesced', 'wake_fallback')`,
|
||||
),
|
||||
}),
|
||||
);
|
||||
|
|
@ -1067,6 +1067,9 @@ export type {
|
|||
SuggestTasksResult,
|
||||
AskUserQuestionsQuestionOption,
|
||||
AskUserQuestionsQuestion,
|
||||
PaperclipQuestionSetOption,
|
||||
PaperclipQuestionSetQuestion,
|
||||
PaperclipQuestionSetPayload,
|
||||
AskUserQuestionsPayload,
|
||||
AskUserQuestionsAnswer,
|
||||
AskUserQuestionsResult,
|
||||
|
|
|
|||
|
|
@ -692,6 +692,9 @@ export type {
|
|||
SuggestTasksResult,
|
||||
AskUserQuestionsQuestionOption,
|
||||
AskUserQuestionsQuestion,
|
||||
PaperclipQuestionSetOption,
|
||||
PaperclipQuestionSetQuestion,
|
||||
PaperclipQuestionSetPayload,
|
||||
AskUserQuestionsPayload,
|
||||
AskUserQuestionsAnswer,
|
||||
AskUserQuestionsResult,
|
||||
|
|
|
|||
|
|
@ -1112,12 +1112,60 @@ export interface AskUserQuestionsQuestion {
|
|||
options: AskUserQuestionsQuestionOption[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Provider-neutral presentation retained when a live harness question has to
|
||||
* fall back to the durable issue interaction lifecycle. This intentionally
|
||||
* mirrors `paperclip.question_set.v1` without making the shared package depend
|
||||
* on a particular runner implementation.
|
||||
*/
|
||||
export interface PaperclipQuestionSetOption {
|
||||
id: string;
|
||||
label: string;
|
||||
description?: string;
|
||||
recommended?: boolean;
|
||||
}
|
||||
|
||||
export interface PaperclipQuestionSetQuestion {
|
||||
id: string;
|
||||
header?: string;
|
||||
prompt: string;
|
||||
helpText?: string;
|
||||
required: boolean;
|
||||
answerMode: "single_select" | "multi_select" | "text";
|
||||
options?: PaperclipQuestionSetOption[];
|
||||
customAnswer?: {
|
||||
enabled: true;
|
||||
label?: string;
|
||||
placeholder?: string;
|
||||
};
|
||||
textValidation?: {
|
||||
minLength?: number;
|
||||
maxLength?: number;
|
||||
pattern?: string;
|
||||
inputType?: "text" | "number" | "integer";
|
||||
minimum?: number;
|
||||
maximum?: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface PaperclipQuestionSetPayload {
|
||||
schema: "paperclip.question_set.v1";
|
||||
title?: string;
|
||||
description?: string;
|
||||
submitLabel?: string;
|
||||
questions: PaperclipQuestionSetQuestion[];
|
||||
}
|
||||
|
||||
export interface AskUserQuestionsPayload {
|
||||
version: 1;
|
||||
title?: string | null;
|
||||
submitLabel?: string | null;
|
||||
supersedeOnUserComment?: boolean;
|
||||
questions: AskUserQuestionsQuestion[];
|
||||
/** Exact presentation for a recovered harness request. */
|
||||
questionSet?: PaperclipQuestionSetPayload;
|
||||
/** Correlates a recovered interaction with the live runtime request it replaces. */
|
||||
runtimeRequestId?: string | null;
|
||||
}
|
||||
|
||||
export interface AskUserQuestionsAnswer {
|
||||
|
|
|
|||
|
|
@ -846,12 +846,54 @@ export const askUserQuestionsQuestionSchema = z.object({
|
|||
options: z.array(askUserQuestionsQuestionOptionSchema).min(1).max(10),
|
||||
});
|
||||
|
||||
const paperclipQuestionOptionSchema = z.object({
|
||||
id: z.string().min(1).max(160),
|
||||
label: z.string().min(1).max(1000),
|
||||
description: z.string().max(4000).optional(),
|
||||
recommended: z.boolean().optional(),
|
||||
});
|
||||
|
||||
const paperclipQuestionSchema = z.object({
|
||||
id: z.string().min(1).max(160),
|
||||
header: z.string().max(1000).optional(),
|
||||
prompt: z.string().min(1).max(4000),
|
||||
helpText: z.string().max(4000).optional(),
|
||||
required: z.boolean(),
|
||||
answerMode: z.enum(["single_select", "multi_select", "text"]),
|
||||
options: z.array(paperclipQuestionOptionSchema).max(128).optional(),
|
||||
customAnswer: z.object({
|
||||
enabled: z.literal(true),
|
||||
label: z.string().max(1000).optional(),
|
||||
placeholder: z.string().max(1000).optional(),
|
||||
}).optional(),
|
||||
textValidation: z.object({
|
||||
minLength: z.number().int().min(0).max(100000).optional(),
|
||||
maxLength: z.number().int().min(0).max(100000).optional(),
|
||||
pattern: z.string().max(1000).optional(),
|
||||
inputType: z.enum(["text", "number", "integer"]).optional(),
|
||||
minimum: z.number().finite().optional(),
|
||||
maximum: z.number().finite().optional(),
|
||||
}).optional(),
|
||||
});
|
||||
|
||||
const paperclipQuestionSetSchema = z.object({
|
||||
schema: z.literal("paperclip.question_set.v1"),
|
||||
title: z.string().max(1000).optional(),
|
||||
description: z.string().max(4000).optional(),
|
||||
submitLabel: z.string().max(200).optional(),
|
||||
questions: z.array(paperclipQuestionSchema).min(1).max(64),
|
||||
});
|
||||
|
||||
export const askUserQuestionsPayloadSchema = z.object({
|
||||
version: z.literal(1),
|
||||
title: z.string().trim().max(240).nullable().optional(),
|
||||
submitLabel: z.string().trim().max(120).nullable().optional(),
|
||||
supersedeOnUserComment: z.boolean().optional(),
|
||||
questions: z.array(askUserQuestionsQuestionSchema).min(1).max(10),
|
||||
/** Exact canonical presentation retained for a recovered harness request. */
|
||||
questionSet: paperclipQuestionSetSchema.optional(),
|
||||
/** Stable correlation for draft handoff from a live runtime request. */
|
||||
runtimeRequestId: z.string().trim().min(1).max(255).nullable().optional(),
|
||||
}).superRefine((value, ctx) => {
|
||||
const seenQuestionIds = new Set<string>();
|
||||
for (const [questionIndex, question] of value.questions.entries()) {
|
||||
|
|
|
|||
|
|
@ -47,6 +47,9 @@ const mockInteractionService = vi.hoisted(() => ({
|
|||
const mockHeartbeatService = vi.hoisted(() => ({
|
||||
wakeup: vi.fn(async () => undefined),
|
||||
}));
|
||||
const mockQuestionResponseDeliveries = vi.hoisted(() => ({
|
||||
deliver: vi.fn(async () => null),
|
||||
}));
|
||||
const mockResolveTaskWatchdogMutationScope = vi.hoisted(() => vi.fn(async () => ({ kind: "none" })));
|
||||
const mockResolveCoreTrustPreset = vi.hoisted(() => vi.fn(() => ({ kind: "standard" })));
|
||||
const mockRunAttribution = vi.hoisted(() => ({
|
||||
|
|
@ -145,6 +148,9 @@ vi.mock("../services/trust-preset-resolver.js", () => ({
|
|||
}));
|
||||
|
||||
function registerModuleMocks() {
|
||||
vi.doMock("../services/question-response-delivery.js", () => ({
|
||||
questionResponseDeliveryService: () => mockQuestionResponseDeliveries,
|
||||
}));
|
||||
vi.doMock("../services/index.js", () => ({
|
||||
companyService: () => ({
|
||||
getById: vi.fn(async () => ({ id: "company-1", attachmentMaxBytes: 10 * 1024 * 1024 })),
|
||||
|
|
@ -283,6 +289,7 @@ describe.sequential("issue thread interaction routes", () => {
|
|||
registerModuleMocks();
|
||||
vi.clearAllMocks();
|
||||
mockInteractionService.getForIssue.mockReset();
|
||||
mockQuestionResponseDeliveries.deliver.mockResolvedValue(null);
|
||||
mockResolveTaskWatchdogMutationScope.mockReset();
|
||||
mockResolveCoreTrustPreset.mockReset();
|
||||
mockAccessDecide.mockReset();
|
||||
|
|
@ -749,7 +756,7 @@ describe.sequential("issue thread interaction routes", () => {
|
|||
);
|
||||
});
|
||||
|
||||
it("answers questions and emits a continuation wake", async () => {
|
||||
it("answers questions through the durable delivery service", async () => {
|
||||
const app = await createApp();
|
||||
|
||||
const res = await request(app)
|
||||
|
|
@ -760,19 +767,10 @@ describe.sequential("issue thread interaction routes", () => {
|
|||
|
||||
expect(res.status).toBe(200);
|
||||
expect(mockInteractionService.answerQuestions).toHaveBeenCalled();
|
||||
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
|
||||
ASSIGNEE_AGENT_ID,
|
||||
expect.objectContaining({
|
||||
reason: "issue_commented",
|
||||
payload: expect.objectContaining({
|
||||
interactionId: "interaction-2",
|
||||
interactionKind: "ask_user_questions",
|
||||
interactionStatus: "answered",
|
||||
sourceCommentId: "comment-2",
|
||||
sourceRunId: RUN_2,
|
||||
}),
|
||||
}),
|
||||
expect(mockQuestionResponseDeliveries.deliver).toHaveBeenCalledWith(
|
||||
"interaction-2",
|
||||
);
|
||||
expect(mockHeartbeatService.wakeup).not.toHaveBeenCalled();
|
||||
expect(mockLogActivity).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
expect.objectContaining({
|
||||
|
|
@ -1902,10 +1900,8 @@ describe.sequential("issue thread interaction routes", () => {
|
|||
expect.anything(),
|
||||
expect.objectContaining({ agentId: ASSIGNEE_AGENT_ID, runId: RUN_2, userId: null }),
|
||||
);
|
||||
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
|
||||
ASSIGNEE_AGENT_ID,
|
||||
expect.objectContaining({ idempotencyKey: "interaction:interaction-2:answered" }),
|
||||
);
|
||||
expect(mockQuestionResponseDeliveries.deliver).toHaveBeenCalledWith("interaction-2");
|
||||
expect(mockHeartbeatService.wakeup).not.toHaveBeenCalled();
|
||||
expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({
|
||||
actorType: "agent",
|
||||
agentId: ASSIGNEE_AGENT_ID,
|
||||
|
|
@ -2327,6 +2323,12 @@ describe.sequential("issue thread interaction routes", () => {
|
|||
effectiveResolverPolicy: "board_or_agents",
|
||||
payload: { version: 1, questions: [] },
|
||||
};
|
||||
mockInteractionService.answerQuestions.mockImplementationOnce(async (_issue, interactionId) => ({
|
||||
...addressed,
|
||||
id: interactionId,
|
||||
status: "answered",
|
||||
result: { version: 1, answers: [] },
|
||||
}));
|
||||
mockInteractionService.getForIssue
|
||||
.mockResolvedValueOnce(addressed)
|
||||
.mockResolvedValueOnce(addressed)
|
||||
|
|
@ -2346,10 +2348,8 @@ describe.sequential("issue thread interaction routes", () => {
|
|||
.post("/api/issues/aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa/interactions/interaction-addressed/respond")
|
||||
.send({ answers: [] });
|
||||
expect(addressee.status).toBe(200);
|
||||
expect(mockHeartbeatService.wakeup).toHaveBeenCalledWith(
|
||||
ASSIGNEE_AGENT_ID,
|
||||
expect.objectContaining({ idempotencyKey: "interaction:interaction-2:answered" }),
|
||||
);
|
||||
expect(mockQuestionResponseDeliveries.deliver).toHaveBeenCalledWith("interaction-addressed");
|
||||
expect(mockHeartbeatService.wakeup).not.toHaveBeenCalled();
|
||||
|
||||
const unrelatedApp = await createApp({
|
||||
type: "agent",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,727 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
goals,
|
||||
heartbeatRuns,
|
||||
issueQuestionResponseDeliveries,
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
|
||||
import {
|
||||
buildQuestionResponseDeliveryEnvelope,
|
||||
formatQuestionResponseSteeringMessage,
|
||||
questionResponseDeliveryService,
|
||||
} from "../services/question-response-delivery.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
const DIRECT_ADAPTER_TYPES = [
|
||||
"acpx_local",
|
||||
"claude_local",
|
||||
"codex_local",
|
||||
"cursor_cloud",
|
||||
"cursor",
|
||||
"gemini_local",
|
||||
"grok_local",
|
||||
"hermes_gateway",
|
||||
"hermes_local",
|
||||
"kimi_local",
|
||||
"openclaw_gateway",
|
||||
"opencode_local",
|
||||
"pi_local",
|
||||
"process",
|
||||
"http",
|
||||
"external_test_adapter",
|
||||
] as const;
|
||||
|
||||
describeEmbeddedPostgres("question response delivery", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-question-delivery-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(issueQuestionResponseDeliveries);
|
||||
await db.delete(issueThreadInteractions);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agentWakeupRequests);
|
||||
await db.delete(issues);
|
||||
await db.delete(goals);
|
||||
await db.delete(agents);
|
||||
await db.delete(companies);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
async function seed(args: {
|
||||
adapterType?: string;
|
||||
runtimeMode?: "legacy" | "native";
|
||||
sourceStatus?: string;
|
||||
successorStatus?: "queued" | "running";
|
||||
} = {}) {
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const goalId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
const sourceRunId = randomUUID();
|
||||
const successorRunId = args.successorStatus ? randomUUID() : null;
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Question delivery",
|
||||
issuePrefix: `Q${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Runner",
|
||||
role: "engineer",
|
||||
status: "active",
|
||||
adapterType: args.adapterType ?? "codex_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(goals).values({ id: goalId, companyId, title: "Test", level: "task", status: "active" });
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
goalId,
|
||||
title: "Deliver answers",
|
||||
status: "in_progress",
|
||||
priority: "medium",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: sourceRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "manual",
|
||||
status: args.sourceStatus ?? "succeeded",
|
||||
runtimeMode: args.runtimeMode ?? "native",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId },
|
||||
...(args.sourceStatus === "running" ? { startedAt: new Date() } : { finishedAt: new Date() }),
|
||||
});
|
||||
if (successorRunId && args.successorStatus) {
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: successorRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
invocationSource: "manual",
|
||||
status: args.successorStatus,
|
||||
runtimeMode: args.runtimeMode ?? "native",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId },
|
||||
...(args.successorStatus === "running" ? { startedAt: new Date() } : {}),
|
||||
});
|
||||
}
|
||||
|
||||
const interactionSvc = issueThreadInteractionService(db);
|
||||
const interaction = await interactionSvc.create(
|
||||
{ id: issueId, companyId },
|
||||
{
|
||||
kind: "ask_user_questions",
|
||||
continuationPolicy: "wake_assignee",
|
||||
sourceRunId,
|
||||
payload: {
|
||||
version: 1,
|
||||
title: "Server choices",
|
||||
questions: [
|
||||
{ id: "purpose", prompt: "What is it for?", selectionMode: "single", required: true, options: [{ id: "custom", label: "Write an answer", freeText: true }] },
|
||||
{ id: "runtime", prompt: "Which runtime?", selectionMode: "single", required: true, options: [{ id: "node", label: "Node.js" }, { id: "bun", label: "Bun" }] },
|
||||
{ id: "features", prompt: "Which features?", selectionMode: "multi", options: [{ id: "health", label: "Health check" }, { id: "logs", label: "Request logs" }] },
|
||||
],
|
||||
questionSet: {
|
||||
schema: "paperclip.question_set.v1",
|
||||
title: "Server choices",
|
||||
questions: [
|
||||
{ id: "purpose", header: "Purpose", prompt: "What is it for?", required: true, answerMode: "text" },
|
||||
{ id: "runtime", header: "Runtime", prompt: "Which runtime?", required: true, answerMode: "single_select", options: [{ id: "node", label: "Node.js" }, { id: "bun", label: "Bun" }] },
|
||||
{ id: "features", header: "Features", prompt: "Which features?", required: false, answerMode: "multi_select", options: [{ id: "health", label: "Health check" }, { id: "logs", label: "Request logs" }], customAnswer: { enabled: true, label: "Other" } },
|
||||
],
|
||||
},
|
||||
},
|
||||
},
|
||||
{ agentId, runId: sourceRunId },
|
||||
);
|
||||
const answered = await interactionSvc.answerQuestions(
|
||||
{ id: issueId, companyId, status: "in_progress" },
|
||||
interaction.id,
|
||||
{ answers: [
|
||||
{ questionId: "purpose", optionIds: [], otherText: "Internal API" },
|
||||
{ questionId: "runtime", optionIds: ["node"] },
|
||||
{ questionId: "features", optionIds: ["health", "logs"], otherText: "Metrics" },
|
||||
] },
|
||||
{ userId: "board-user" },
|
||||
);
|
||||
return { companyId, agentId, issueId, sourceRunId, successorRunId, interaction: answered };
|
||||
}
|
||||
|
||||
it("persists the receipt atomically and steers exactly once into a running successor", async () => {
|
||||
const seeded = await seed({ successorStatus: "running" });
|
||||
const newerRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: newerRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "manual",
|
||||
status: "running",
|
||||
runtimeMode: "native",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
startedAt: new Date(),
|
||||
});
|
||||
await db.update(issues).set({ executionRunId: seeded.successorRunId })
|
||||
.where(eq(issues.id, seeded.issueId));
|
||||
const persistedBeforeDelivery = await db.select().from(issueQuestionResponseDeliveries)
|
||||
.where(eq(issueQuestionResponseDeliveries.interactionId, seeded.interaction.id))
|
||||
.then((rows) => rows[0]);
|
||||
expect(persistedBeforeDelivery).toMatchObject({
|
||||
status: "pending",
|
||||
correlationId: `question-response:${seeded.interaction.id}`,
|
||||
sourceRunId: seeded.sourceRunId,
|
||||
});
|
||||
|
||||
const steer = vi.fn().mockResolvedValue({ turnId: "turn-successor" });
|
||||
const wakeup = vi.fn();
|
||||
const service = questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer,
|
||||
});
|
||||
const first = await service.deliver(seeded.interaction.id);
|
||||
const second = await service.deliver(seeded.interaction.id);
|
||||
|
||||
expect(first).toMatchObject({
|
||||
status: "delivered",
|
||||
mode: "steered",
|
||||
targetRunId: seeded.successorRunId,
|
||||
targetTurnId: "turn-successor",
|
||||
duplicate: false,
|
||||
});
|
||||
expect(second).toMatchObject({ mode: "steered", duplicate: true });
|
||||
expect(steer).toHaveBeenCalledTimes(1);
|
||||
expect(steer).toHaveBeenCalledWith(expect.objectContaining({
|
||||
runId: seeded.successorRunId,
|
||||
correlationId: `question-response:${seeded.interaction.id}`,
|
||||
message: expect.stringContaining("- Runtime — Which runtime?: Node.js"),
|
||||
}));
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
|
||||
const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivery).toMatchObject({
|
||||
status: "delivered",
|
||||
deliveryMode: "steered",
|
||||
targetRunId: seeded.successorRunId,
|
||||
targetTurnId: "turn-successor",
|
||||
attemptCount: 1,
|
||||
});
|
||||
const deliveryEvents = await db.select().from(activityLog)
|
||||
.where(eq(activityLog.action, "issue.question_response_delivered"));
|
||||
expect(deliveryEvents).toHaveLength(1);
|
||||
expect(JSON.stringify(deliveryEvents[0]?.details)).not.toContain("Internal API");
|
||||
expect(JSON.stringify(deliveryEvents[0]?.details)).not.toContain("Node.js");
|
||||
});
|
||||
|
||||
it("coalesces into a queued successor without creating another wake", async () => {
|
||||
const seeded = await seed({ successorStatus: "queued" });
|
||||
const successor = await db.select().from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, seeded.successorRunId!))
|
||||
.then((rows) => rows[0]!);
|
||||
const wakeup = vi.fn().mockResolvedValue(successor);
|
||||
const steer = vi.fn();
|
||||
const outcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer,
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(outcome).toMatchObject({ status: "delivered", mode: "coalesced", targetRunId: successor.id });
|
||||
expect(steer).not.toHaveBeenCalled();
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
expect(wakeup.mock.calls[0]?.[1]).toMatchObject({
|
||||
idempotencyKey: `question-response:${seeded.interaction.id}`,
|
||||
contextSnapshot: {
|
||||
interactionId: seeded.interaction.id,
|
||||
interactionStatus: "answered",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("never steers into the source run and keeps a skipped wake retryable", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const wakeup = vi.fn().mockResolvedValue(null);
|
||||
const steer = vi.fn();
|
||||
const outcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer,
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(outcome).toBeNull();
|
||||
expect(steer).not.toHaveBeenCalled();
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
expect(JSON.stringify(wakeup.mock.calls[0])).not.toContain("Internal API");
|
||||
expect(JSON.stringify(wakeup.mock.calls[0])).not.toContain("Node.js");
|
||||
const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivery).toMatchObject({
|
||||
status: "pending",
|
||||
attemptCount: 1,
|
||||
lastErrorCode: "question_response_wake_skipped",
|
||||
});
|
||||
});
|
||||
|
||||
it("delivers after wake suppression outlasts the bounded error retry limit", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const fallbackRunId = randomUUID();
|
||||
let wakeAttempts = 0;
|
||||
const wakeup = vi.fn().mockImplementation(async () => {
|
||||
wakeAttempts += 1;
|
||||
if (wakeAttempts <= 5) return null;
|
||||
return db.insert(heartbeatRuns).values({
|
||||
id: fallbackRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
runtimeMode: "legacy",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
}).returning().then((rows) => rows[0]!);
|
||||
});
|
||||
const service = questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer: vi.fn(),
|
||||
});
|
||||
|
||||
for (let attempt = 0; attempt < 5; attempt += 1) {
|
||||
await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull();
|
||||
}
|
||||
const delivered = await service.deliver(seeded.interaction.id);
|
||||
|
||||
expect(delivered).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
mode: "wake_fallback",
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivery).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
attemptCount: 6,
|
||||
errorCount: 0,
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves the actual-error retry budget after prolonged wake suppression", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const fallbackRunId = randomUUID();
|
||||
let wakeAttempts = 0;
|
||||
const wakeup = vi.fn().mockImplementation(async () => {
|
||||
wakeAttempts += 1;
|
||||
if (wakeAttempts <= 5) return null;
|
||||
if (wakeAttempts === 6) throw new Error("scheduler temporarily unavailable");
|
||||
return db.insert(heartbeatRuns).values({
|
||||
id: fallbackRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
runtimeMode: "legacy",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
}).returning().then((rows) => rows[0]!);
|
||||
});
|
||||
const service = questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer: vi.fn(),
|
||||
});
|
||||
|
||||
for (let attempt = 0; attempt < 5; attempt += 1) {
|
||||
await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull();
|
||||
}
|
||||
await expect(service.deliver(seeded.interaction.id)).resolves.toBeNull();
|
||||
const [afterError] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(afterError).toMatchObject({
|
||||
status: "pending",
|
||||
attemptCount: 6,
|
||||
errorCount: 1,
|
||||
lastErrorCode: "scheduler temporarily unavailable",
|
||||
});
|
||||
|
||||
await expect(service.deliver(seeded.interaction.id)).resolves.toMatchObject({
|
||||
status: "fallback_queued",
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
const [delivered] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivered).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
attemptCount: 7,
|
||||
errorCount: 1,
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
});
|
||||
|
||||
it("enforces one durable wake per question-response idempotency key", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const idempotencyKey = `question-response:${seeded.interaction.id}`;
|
||||
const request = {
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
idempotencyKey,
|
||||
} as const;
|
||||
|
||||
await db.insert(agentWakeupRequests).values({ ...request, status: "queued" });
|
||||
await expect(db.insert(agentWakeupRequests).values({
|
||||
...request,
|
||||
status: "coalesced",
|
||||
})).rejects.toMatchObject({ cause: { code: "23505" } });
|
||||
|
||||
// Suppression receipts are intentionally outside the fence so the outbox
|
||||
// can retry after scheduling is enabled again.
|
||||
await expect(db.insert(agentWakeupRequests).values({
|
||||
...request,
|
||||
status: "skipped",
|
||||
finishedAt: new Date(),
|
||||
})).resolves.toBeDefined();
|
||||
});
|
||||
|
||||
it("reuses the winning wake when a concurrent insert hits the idempotency fence", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const fallbackRunId = randomUUID();
|
||||
const wakeup = vi.fn().mockImplementation(async (
|
||||
_agentId: string,
|
||||
options: { idempotencyKey?: string | null },
|
||||
) => {
|
||||
const request = {
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
idempotencyKey: options.idempotencyKey,
|
||||
} as const;
|
||||
const [winner] = await db.insert(agentWakeupRequests).values({
|
||||
...request,
|
||||
status: "queued",
|
||||
}).returning();
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: fallbackRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
runtimeMode: "legacy",
|
||||
driverKind: "codex",
|
||||
wakeupRequestId: winner!.id,
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
});
|
||||
await db.update(agentWakeupRequests).set({ runId: fallbackRunId })
|
||||
.where(eq(agentWakeupRequests.id, winner!.id));
|
||||
|
||||
// Model the losing claimant reaching the same transactional insert after
|
||||
// the winner commits. The service must recover the winner's receipt.
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
...request,
|
||||
status: "coalesced",
|
||||
});
|
||||
throw new Error("unreachable");
|
||||
});
|
||||
|
||||
const outcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer: vi.fn(),
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(outcome).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
mode: "wake_fallback",
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivery).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
attemptCount: 1,
|
||||
errorCount: 0,
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
});
|
||||
|
||||
it("reuses a durable wake receipt instead of issuing a duplicate continuation", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const [wakeRequest] = await db.insert(agentWakeupRequests).values({
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
status: "queued",
|
||||
idempotencyKey: `question-response:${seeded.interaction.id}`,
|
||||
}).returning();
|
||||
const [wakeRun] = await db.insert(heartbeatRuns).values({
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
runtimeMode: "legacy",
|
||||
wakeupRequestId: wakeRequest!.id,
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
}).returning();
|
||||
await db.update(agentWakeupRequests).set({ runId: wakeRun!.id })
|
||||
.where(eq(agentWakeupRequests.id, wakeRequest!.id));
|
||||
const wakeup = vi.fn();
|
||||
|
||||
const outcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer: vi.fn(),
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(outcome).toMatchObject({
|
||||
status: "delivered",
|
||||
mode: "coalesced",
|
||||
targetRunId: wakeRun!.id,
|
||||
});
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("recovers a completed wake when receipt finalization was interrupted", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
const [wakeRequest] = await db.insert(agentWakeupRequests).values({
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
status: "completed",
|
||||
idempotencyKey: `question-response:${seeded.interaction.id}`,
|
||||
finishedAt: new Date(),
|
||||
}).returning();
|
||||
const [wakeRun] = await db.insert(heartbeatRuns).values({
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "succeeded",
|
||||
runtimeMode: "legacy",
|
||||
wakeupRequestId: wakeRequest!.id,
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
finishedAt: new Date(),
|
||||
}).returning();
|
||||
await db.update(agentWakeupRequests).set({ runId: wakeRun!.id })
|
||||
.where(eq(agentWakeupRequests.id, wakeRequest!.id));
|
||||
const wakeup = vi.fn();
|
||||
|
||||
const outcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer: vi.fn(),
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(outcome).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
mode: "wake_fallback",
|
||||
targetRunId: wakeRun!.id,
|
||||
});
|
||||
expect(wakeup).not.toHaveBeenCalled();
|
||||
const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivery).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
errorCount: 0,
|
||||
targetRunId: wakeRun!.id,
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps a long wake claim leased while the side effect is active", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
let releaseWake!: (value: null) => void;
|
||||
const wakeup = vi.fn(() => new Promise<null>((resolve) => {
|
||||
releaseWake = resolve;
|
||||
}));
|
||||
const service = questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer: vi.fn(),
|
||||
claimStaleMs: 40,
|
||||
claimRefreshMs: 5,
|
||||
});
|
||||
|
||||
const deliveryPromise = service.deliver(seeded.interaction.id);
|
||||
await vi.waitFor(() => expect(wakeup).toHaveBeenCalledTimes(1));
|
||||
await new Promise((resolve) => setTimeout(resolve, 70));
|
||||
await expect(service.sweepPending()).resolves.toMatchObject({ scanned: 0 });
|
||||
releaseWake(null);
|
||||
await expect(deliveryPromise).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("fences a stale worker after a newer claim generation takes ownership", async () => {
|
||||
const seeded = await seed({ sourceStatus: "running" });
|
||||
let releaseFirstWake!: (value: { id: string; driverKind: string }) => void;
|
||||
const firstWakeup = vi.fn(() => new Promise<{ id: string; driverKind: string }>((resolve) => {
|
||||
releaseFirstWake = resolve;
|
||||
}));
|
||||
const firstService = questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup: firstWakeup } as never,
|
||||
steer: vi.fn(),
|
||||
claimStaleMs: 40,
|
||||
claimRefreshMs: 5,
|
||||
});
|
||||
|
||||
const firstDelivery = firstService.deliver(seeded.interaction.id);
|
||||
await vi.waitFor(() => expect(firstWakeup).toHaveBeenCalledTimes(1));
|
||||
|
||||
// Simulate recovery after the first worker stopped renewing. The next
|
||||
// claim increments attemptCount, which is the fencing generation.
|
||||
await db.update(issueQuestionResponseDeliveries).set({
|
||||
status: "pending",
|
||||
lastAttemptAt: new Date(0),
|
||||
}).where(eq(issueQuestionResponseDeliveries.interactionId, seeded.interaction.id));
|
||||
|
||||
const secondRunId = randomUUID();
|
||||
const secondWakeup = vi.fn().mockImplementation(async () => db.insert(heartbeatRuns).values({
|
||||
id: secondRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
runtimeMode: "legacy",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
}).returning().then((rows) => rows[0]!));
|
||||
const secondOutcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup: secondWakeup } as never,
|
||||
steer: vi.fn(),
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(secondOutcome).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
targetRunId: secondRunId,
|
||||
duplicate: false,
|
||||
});
|
||||
releaseFirstWake({ id: randomUUID(), driverKind: "codex" });
|
||||
await expect(firstDelivery).resolves.toMatchObject({
|
||||
status: "fallback_queued",
|
||||
targetRunId: secondRunId,
|
||||
duplicate: true,
|
||||
});
|
||||
|
||||
const [delivery] = await db.select().from(issueQuestionResponseDeliveries);
|
||||
expect(delivery).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
targetRunId: secondRunId,
|
||||
attemptCount: 2,
|
||||
});
|
||||
const deliveryEvents = await db.select().from(activityLog)
|
||||
.where(eq(activityLog.action, "issue.question_response_delivered"));
|
||||
expect(deliveryEvents).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(DIRECT_ADAPTER_TYPES)(
|
||||
"keeps %s on the existing wake path without invoking native steering",
|
||||
async (adapterType) => {
|
||||
const seeded = await seed({
|
||||
adapterType,
|
||||
runtimeMode: "legacy",
|
||||
successorStatus: "running",
|
||||
});
|
||||
const fallbackRunId = randomUUID();
|
||||
const wakeup = vi.fn().mockImplementation(async () => db.insert(heartbeatRuns).values({
|
||||
id: fallbackRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
runtimeMode: "legacy",
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
}).returning().then((rows) => rows[0]!));
|
||||
const steer = vi.fn();
|
||||
|
||||
const outcome = await questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer,
|
||||
}).deliver(seeded.interaction.id);
|
||||
|
||||
expect(outcome).toMatchObject({
|
||||
status: "fallback_queued",
|
||||
mode: "wake_fallback",
|
||||
targetRunId: fallbackRunId,
|
||||
});
|
||||
expect(steer).not.toHaveBeenCalled();
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
expect(wakeup.mock.calls[0]?.[1]).toMatchObject({
|
||||
idempotencyKey: `question-response:${seeded.interaction.id}`,
|
||||
contextSnapshot: {
|
||||
issueId: seeded.issueId,
|
||||
interactionId: seeded.interaction.id,
|
||||
interactionStatus: "answered",
|
||||
},
|
||||
});
|
||||
},
|
||||
);
|
||||
|
||||
it("falls back once when successor steering is unsupported", async () => {
|
||||
const seeded = await seed({ successorStatus: "running" });
|
||||
const fallbackRunId = randomUUID();
|
||||
const steer = vi.fn().mockRejectedValue(
|
||||
Object.assign(new Error("unsupported"), {
|
||||
code: "steering_unsupported",
|
||||
}),
|
||||
);
|
||||
const wakeup = vi.fn().mockImplementation(async () => db.insert(heartbeatRuns).values({
|
||||
id: fallbackRunId,
|
||||
companyId: seeded.companyId,
|
||||
agentId: seeded.agentId,
|
||||
invocationSource: "automation",
|
||||
status: "queued",
|
||||
driverKind: "codex",
|
||||
contextSnapshot: { issueId: seeded.issueId },
|
||||
}).returning().then((rows) => rows[0]!));
|
||||
const service = questionResponseDeliveryService(db, {
|
||||
heartbeat: { wakeup } as never,
|
||||
steer,
|
||||
});
|
||||
const first = await service.deliver(seeded.interaction.id);
|
||||
const second = await service.deliver(seeded.interaction.id);
|
||||
|
||||
expect(first).toMatchObject({ status: "fallback_queued", mode: "wake_fallback", targetRunId: fallbackRunId });
|
||||
expect(second?.duplicate).toBe(true);
|
||||
expect(steer).toHaveBeenCalledTimes(1);
|
||||
expect(wakeup).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("formats text, select labels, multi-select, and custom answers in order", async () => {
|
||||
const seeded = await seed();
|
||||
const envelope = buildQuestionResponseDeliveryEnvelope(seeded.interaction);
|
||||
expect(envelope.response).toEqual({
|
||||
schema: "paperclip.question_response.v1",
|
||||
answers: {
|
||||
purpose: { text: "Internal API" },
|
||||
runtime: { selectedOptionIds: ["node"] },
|
||||
features: { selectedOptionIds: ["health", "logs"], customText: "Metrics" },
|
||||
},
|
||||
});
|
||||
expect(formatQuestionResponseSteeringMessage(envelope)).toBe([
|
||||
"Answered questions",
|
||||
"",
|
||||
"- Purpose — What is it for?: Internal API",
|
||||
"- Runtime — Which runtime?: Node.js",
|
||||
"- Features — Which features?: Health check, Request logs, Metrics",
|
||||
].join("\n"));
|
||||
});
|
||||
});
|
||||
|
|
@ -314,6 +314,18 @@ vi.mock("../services/index.js", () => ({
|
|||
})),
|
||||
}));
|
||||
|
||||
vi.mock("../services/question-response-delivery.js", () => ({
|
||||
questionResponseDeliveryService: vi.fn(() => ({
|
||||
sweepPending: vi.fn(async () => ({
|
||||
scanned: 0,
|
||||
steered: 0,
|
||||
coalesced: 0,
|
||||
wakeFallback: 0,
|
||||
failed: 0,
|
||||
})),
|
||||
})),
|
||||
}));
|
||||
|
||||
vi.mock("../services/secret-proposals.js", () => ({
|
||||
createSecretProposalsService: vi.fn(() => ({
|
||||
sweepExpired: vi.fn(async () => 0),
|
||||
|
|
|
|||
|
|
@ -67,6 +67,7 @@ import {
|
|||
toolAccessService,
|
||||
workspaceOperationService,
|
||||
} from "./services/index.js";
|
||||
import { questionResponseDeliveryService } from "./services/question-response-delivery.js";
|
||||
import { queueIssueAssignmentWakeup } from "./services/issue-assignment-wakeup.js";
|
||||
import { createSecretProposalsService } from "./services/secret-proposals.js";
|
||||
import { environmentRuntimeService } from "./services/environment-runtime.js";
|
||||
|
|
@ -1116,6 +1117,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 })
|
||||
|
|
@ -1132,6 +1136,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);
|
||||
|
|
@ -1542,6 +1554,16 @@ 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
|
||||
|
|
|
|||
|
|
@ -140,6 +140,7 @@ import {
|
|||
routineService,
|
||||
workProductService,
|
||||
} from "../services/index.js";
|
||||
import { questionResponseDeliveryService } from "../services/question-response-delivery.js";
|
||||
import { artifactReviewDocumentService } from "../services/artifact-review-documents.js";
|
||||
import { assertCanResolveProposal } from "../services/secret-proposal-authorization.js";
|
||||
import { buildDocumentReviewContext, buildPlanReviewContext } from "../services/plan-review-context.js";
|
||||
|
|
@ -2839,6 +2840,9 @@ export function issueRoutes(
|
|||
const decisionTrainingSvc = decisionTrainingService(db);
|
||||
const issueReferencesSvc = issueReferenceService(db);
|
||||
const issueThreadInteractionsSvc = issueThreadInteractionService(db);
|
||||
const questionResponseDeliveries = questionResponseDeliveryService(db, {
|
||||
heartbeat,
|
||||
});
|
||||
const memoizeIssueRead = createRequestPromiseMemo<Request, Awaited<ReturnType<typeof svc.getById>>>({
|
||||
shouldCache: (issue) => issue !== null,
|
||||
});
|
||||
|
|
@ -11694,13 +11698,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);
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@ function createFakeDb(args: {
|
|||
const issueTouches: Array<Record<string, unknown>> = [];
|
||||
const interactionUpdates: Array<Record<string, unknown>> = [];
|
||||
const toolActionRequestUpdates: Array<Record<string, unknown>> = [];
|
||||
const inserts: Array<{ table: string; values: Record<string, unknown> }> = [];
|
||||
let selectCallCount = 0;
|
||||
|
||||
const db: any = {
|
||||
|
|
@ -66,7 +67,11 @@ function createFakeDb(args: {
|
|||
};
|
||||
},
|
||||
})),
|
||||
insert: vi.fn(),
|
||||
insert: vi.fn((table: unknown) => ({
|
||||
values: async (values: Record<string, unknown>) => {
|
||||
inserts.push({ table: getTableName(table as never), values });
|
||||
},
|
||||
})),
|
||||
transaction: async (callback: (tx: typeof db) => Promise<void>) => callback(db),
|
||||
};
|
||||
|
||||
|
|
@ -76,6 +81,7 @@ function createFakeDb(args: {
|
|||
issueTouches,
|
||||
interactionUpdates,
|
||||
toolActionRequestUpdates,
|
||||
inserts,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -262,6 +268,16 @@ describe("issueThreadInteractionService", () => {
|
|||
});
|
||||
expect(state.interactionUpdates).toHaveLength(1);
|
||||
expect(state.issueTouches).toHaveLength(1);
|
||||
expect(state.inserts).toEqual([
|
||||
expect.objectContaining({
|
||||
table: "issue_question_response_deliveries",
|
||||
values: expect.objectContaining({
|
||||
interactionId: "interaction-2",
|
||||
correlationId: "question-response:interaction-2",
|
||||
payloadSha256: expect.any(String),
|
||||
}),
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("withdraws a pending interaction with attribution and rejects repeats", async () => {
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -3337,30 +3339,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);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,756 @@
|
|||
import { and, asc, desc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
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 { isUniqueViolation } from "../db-errors.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";
|
||||
|
||||
const DELIVERY_CLAIM_STALE_MS = 30_000;
|
||||
const DELIVERY_CLAIM_REFRESH_MS = 10_000;
|
||||
const MAX_DELIVERY_ATTEMPTS = 5;
|
||||
const DELIVERY_CORRELATION_PREFIX = "question-response:";
|
||||
const QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT =
|
||||
"agent_wakeup_requests_question_response_delivery_idempotency_uq";
|
||||
|
||||
class DeliveryClaimUnavailableError extends Error {
|
||||
constructor() {
|
||||
super("question_response_delivery_claim_unavailable");
|
||||
this.name = "DeliveryClaimUnavailableError";
|
||||
}
|
||||
}
|
||||
const DURABLE_WAKE_REQUEST_STATUSES = [
|
||||
"queued",
|
||||
"claimed",
|
||||
"running",
|
||||
"succeeded",
|
||||
"completed",
|
||||
"coalesced",
|
||||
"deferred_issue_execution",
|
||||
"retrying",
|
||||
"scheduled_retry",
|
||||
] as const;
|
||||
|
||||
type QuestionInteractionRow = typeof issueThreadInteractions.$inferSelect;
|
||||
type DeliveryRow = typeof issueQuestionResponseDeliveries.$inferSelect;
|
||||
type Heartbeat = Pick<ReturnType<typeof heartbeatService>, "wakeup">;
|
||||
type QuestionResponseSteer = (input: {
|
||||
runId: string;
|
||||
message: string;
|
||||
correlationId: string;
|
||||
}) => Promise<{ turnId?: string | null }>;
|
||||
|
||||
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;
|
||||
/** Optional native steering seam. Direct adapters use the durable wake fallback. */
|
||||
steer?: QuestionResponseSteer;
|
||||
now?: () => Date;
|
||||
/** Test-only lease timings. Production callers use the bounded defaults. */
|
||||
claimStaleMs?: number;
|
||||
claimRefreshMs?: number;
|
||||
}
|
||||
|
||||
function readSteeringErrorCode(error: unknown): string {
|
||||
if (
|
||||
error &&
|
||||
typeof error === "object" &&
|
||||
"code" in error &&
|
||||
typeof error.code === "string"
|
||||
) {
|
||||
return error.code;
|
||||
}
|
||||
return "steering_rejected";
|
||||
}
|
||||
|
||||
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;
|
||||
const now = options.now ?? (() => new Date());
|
||||
const claimStaleMs = Math.max(2, options.claimStaleMs ?? DELIVERY_CLAIM_STALE_MS);
|
||||
const claimRefreshMs = Math.max(
|
||||
1,
|
||||
Math.min(options.claimRefreshMs ?? DELIVERY_CLAIM_REFRESH_MS, Math.floor(claimStaleMs / 2)),
|
||||
);
|
||||
|
||||
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() - claimStaleMs
|
||||
) 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.transaction(async (tx) => {
|
||||
const row = await tx.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"),
|
||||
eq(issueQuestionResponseDeliveries.attemptCount, input.delivery.attemptCount),
|
||||
)).returning().then((rows) => rows[0] ?? null);
|
||||
if (!row) return null;
|
||||
await logActivity(tx as unknown as 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: row.id,
|
||||
interactionId: input.interaction.id,
|
||||
sourceRunId: input.interaction.sourceRunId,
|
||||
targetRunId: input.targetRunId,
|
||||
targetTurnId: input.targetTurnId ?? null,
|
||||
correlationId: row.correlationId,
|
||||
payloadSha256: row.payloadSha256,
|
||||
deliveryStatus: input.status,
|
||||
deliveryMode: input.mode,
|
||||
adapter: input.adapter,
|
||||
errorCode: input.errorCode ?? null,
|
||||
},
|
||||
});
|
||||
return row;
|
||||
});
|
||||
|
||||
const persisted = updated ?? await db.select().from(issueQuestionResponseDeliveries)
|
||||
.where(eq(issueQuestionResponseDeliveries.id, input.delivery.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const result: DeliveryRow = persisted ?? input.delivery;
|
||||
if (updated) {
|
||||
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,
|
||||
options: { bounded: boolean } = { bounded: true },
|
||||
) {
|
||||
const at = now();
|
||||
const nextErrorCount = delivery.errorCount + (options.bounded ? 1 : 0);
|
||||
const exhausted = options.bounded && nextErrorCount >= MAX_DELIVERY_ATTEMPTS;
|
||||
await db.update(issueQuestionResponseDeliveries).set({
|
||||
// Keep an exhausted claim owned until recordTerminal commits its outcome.
|
||||
status: exhausted ? "delivering" : "pending",
|
||||
...(options.bounded ? { errorCount: nextErrorCount } : {}),
|
||||
lastErrorCode: errorCode,
|
||||
updatedAt: at,
|
||||
}).where(and(
|
||||
eq(issueQuestionResponseDeliveries.id, delivery.id),
|
||||
eq(issueQuestionResponseDeliveries.status, "delivering"),
|
||||
eq(issueQuestionResponseDeliveries.attemptCount, delivery.attemptCount),
|
||||
));
|
||||
return exhausted;
|
||||
}
|
||||
|
||||
async function withClaimLease<T>(delivery: DeliveryRow, operation: () => Promise<T>): Promise<T> {
|
||||
let stopped = false;
|
||||
let renewal = Promise.resolve();
|
||||
const timer = setInterval(() => {
|
||||
renewal = renewal.then(async () => {
|
||||
if (stopped) return;
|
||||
const renewedAt = now();
|
||||
const renewed = await db.update(issueQuestionResponseDeliveries).set({
|
||||
lastAttemptAt: renewedAt,
|
||||
updatedAt: renewedAt,
|
||||
}).where(and(
|
||||
eq(issueQuestionResponseDeliveries.id, delivery.id),
|
||||
eq(issueQuestionResponseDeliveries.status, "delivering"),
|
||||
eq(issueQuestionResponseDeliveries.attemptCount, delivery.attemptCount),
|
||||
)).returning({ id: issueQuestionResponseDeliveries.id });
|
||||
if (renewed.length === 0) stopped = true;
|
||||
}).catch((error) => {
|
||||
logger.warn({ err: error, deliveryId: delivery.id }, "question response claim lease renewal failed");
|
||||
});
|
||||
}, claimRefreshMs);
|
||||
timer.unref?.();
|
||||
let result: T | undefined;
|
||||
let operationError: unknown;
|
||||
try {
|
||||
result = await operation();
|
||||
} catch (error) {
|
||||
operationError = error;
|
||||
} finally {
|
||||
stopped = true;
|
||||
clearInterval(timer);
|
||||
await renewal;
|
||||
}
|
||||
|
||||
// `attemptCount` is the claim generation. A stale worker must not continue
|
||||
// after a sweep has reclaimed the row for a newer attempt, even if its
|
||||
// external side effect eventually resolves. Confirm ownership after the
|
||||
// side effect so a rejected native steer cannot fall through to a second
|
||||
// wake after losing its claim.
|
||||
let ownsClaim = false;
|
||||
try {
|
||||
ownsClaim = await db.select({ id: issueQuestionResponseDeliveries.id })
|
||||
.from(issueQuestionResponseDeliveries)
|
||||
.where(and(
|
||||
eq(issueQuestionResponseDeliveries.id, delivery.id),
|
||||
eq(issueQuestionResponseDeliveries.status, "delivering"),
|
||||
eq(issueQuestionResponseDeliveries.attemptCount, delivery.attemptCount),
|
||||
))
|
||||
.limit(1)
|
||||
.then((rows) => rows.length === 1);
|
||||
} catch (error) {
|
||||
logger.warn({ err: error, deliveryId: delivery.id }, "question response claim ownership check failed");
|
||||
throw new DeliveryClaimUnavailableError();
|
||||
}
|
||||
if (!ownsClaim) throw new DeliveryClaimUnavailableError();
|
||||
if (operationError !== undefined) throw operationError;
|
||||
return result as T;
|
||||
}
|
||||
|
||||
async function findDurableWakeRequest(input: {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
idempotencyKey: string;
|
||||
}) {
|
||||
const request = await db.select({
|
||||
id: agentWakeupRequests.id,
|
||||
runId: agentWakeupRequests.runId,
|
||||
status: agentWakeupRequests.status,
|
||||
}).from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, input.companyId),
|
||||
eq(agentWakeupRequests.agentId, input.agentId),
|
||||
eq(agentWakeupRequests.idempotencyKey, input.idempotencyKey),
|
||||
inArray(agentWakeupRequests.status, [...DURABLE_WAKE_REQUEST_STATUSES]),
|
||||
)).orderBy(desc(agentWakeupRequests.createdAt)).limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!request?.runId) return request ? { request, run: null } : null;
|
||||
const run = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, request.runId),
|
||||
eq(heartbeatRuns.companyId, input.companyId),
|
||||
eq(heartbeatRuns.agentId, input.agentId),
|
||||
)).limit(1).then((rows) => rows[0] ?? null);
|
||||
return { request, run };
|
||||
}
|
||||
|
||||
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 assigneeAgentId = issue.assigneeAgentId;
|
||||
|
||||
const liveRuns = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, interaction.companyId),
|
||||
eq(heartbeatRuns.agentId, assigneeAgentId),
|
||||
inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
|
||||
)).orderBy(asc(heartbeatRuns.createdAt));
|
||||
const issueRuns = liveRuns.filter((run) => issueIdFromRun(run) === interaction.issueId);
|
||||
// `executionRunId` is the issue's authoritative active-run pointer. Fall
|
||||
// back to the newest matching running row only for legacy/racy rows where
|
||||
// the pointer has not been populated yet; choosing the oldest stale row
|
||||
// could steer an answer into the wrong provider turn.
|
||||
const successorRunning = (
|
||||
issue.executionRunId
|
||||
? issueRuns.find((run) =>
|
||||
run.id === issue.executionRunId &&
|
||||
run.status === "running" &&
|
||||
run.id !== interaction.sourceRunId,
|
||||
)
|
||||
: null
|
||||
) ?? [...issueRuns].reverse().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));
|
||||
if (nativeSha256(envelope) !== claimed.payloadSha256) {
|
||||
return recordTerminal({
|
||||
delivery: claimed,
|
||||
interaction,
|
||||
status: "failed",
|
||||
mode: null,
|
||||
targetRunId: null,
|
||||
adapter,
|
||||
errorCode: "question_response_payload_digest_mismatch",
|
||||
});
|
||||
}
|
||||
|
||||
let steeringErrorCode: string | null = null;
|
||||
if (successorRunning?.runtimeMode === "native" && steer) {
|
||||
try {
|
||||
const acknowledgement = await withClaimLease(claimed, () => 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) {
|
||||
if (error instanceof DeliveryClaimUnavailableError) return terminalOutcome(interactionId);
|
||||
steeringErrorCode = readSteeringErrorCode(error);
|
||||
}
|
||||
} else if (successorRunning) {
|
||||
steeringErrorCode = "steering_unsupported";
|
||||
}
|
||||
|
||||
const actor = actorForInteraction(interaction);
|
||||
// This is a new, migration-fenced namespace. The partial unique index on
|
||||
// agent_wakeup_requests makes the wake transaction itself idempotent, so a
|
||||
// reclaimed stale worker cannot create a second continuation run.
|
||||
const wakeIdempotencyKey = `question-response:${interaction.id}`;
|
||||
try {
|
||||
const existingWake = await findDurableWakeRequest({
|
||||
companyId: interaction.companyId,
|
||||
agentId: assigneeAgentId,
|
||||
idempotencyKey: wakeIdempotencyKey,
|
||||
});
|
||||
const wakeRun = existingWake?.run ?? (existingWake ? null : await withClaimLease(
|
||||
claimed,
|
||||
() => options.heartbeat.wakeup(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: wakeIdempotencyKey,
|
||||
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 durableWake = existingWake ?? (wakeRun ? null : await findDurableWakeRequest({
|
||||
companyId: interaction.companyId,
|
||||
agentId: assigneeAgentId,
|
||||
idempotencyKey: wakeIdempotencyKey,
|
||||
}));
|
||||
if (!wakeRun && !durableWake) {
|
||||
const errorCode = "question_response_wake_skipped";
|
||||
// Scheduling suppression is an availability state, not a delivery
|
||||
// failure. Keep the durable receipt retryable until the suppression is
|
||||
// lifted; the bounded limit remains reserved for actual wake errors.
|
||||
await releaseForRetry(claimed, errorCode, { bounded: false });
|
||||
return null;
|
||||
}
|
||||
const targetRun = wakeRun ?? durableWake?.run ?? queuedSuccessor ?? null;
|
||||
const coalesced = Boolean(queuedSuccessor && targetRun?.id === queuedSuccessor.id);
|
||||
return recordTerminal({
|
||||
delivery: claimed,
|
||||
interaction,
|
||||
status: coalesced ? "delivered" : "fallback_queued",
|
||||
mode: coalesced ? "coalesced" : "wake_fallback",
|
||||
targetRunId: targetRun?.id ?? null,
|
||||
adapter: targetRun?.driverKind ?? adapter,
|
||||
errorCode: steeringErrorCode,
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof DeliveryClaimUnavailableError) return terminalOutcome(interactionId);
|
||||
if (isUniqueViolation(error, QUESTION_RESPONSE_WAKE_IDEMPOTENCY_CONSTRAINT)) {
|
||||
// A concurrent claimant won the transactional wake fence after our
|
||||
// preflight lookup. Reuse its committed receipt instead of consuming
|
||||
// an error retry or issuing another continuation.
|
||||
const durableWake = await findDurableWakeRequest({
|
||||
companyId: interaction.companyId,
|
||||
agentId: assigneeAgentId,
|
||||
idempotencyKey: wakeIdempotencyKey,
|
||||
});
|
||||
if (durableWake) {
|
||||
const targetRun = durableWake.run ?? queuedSuccessor ?? null;
|
||||
const coalesced = Boolean(queuedSuccessor && targetRun?.id === queuedSuccessor.id);
|
||||
return recordTerminal({
|
||||
delivery: claimed,
|
||||
interaction,
|
||||
status: coalesced ? "delivered" : "fallback_queued",
|
||||
mode: coalesced ? "coalesced" : "wake_fallback",
|
||||
targetRunId: targetRun?.id ?? null,
|
||||
adapter: targetRun?.driverKind ?? adapter,
|
||||
errorCode: steeringErrorCode,
|
||||
});
|
||||
}
|
||||
}
|
||||
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,
|
||||
errorCount: claimed.errorCount + 1,
|
||||
exhausted,
|
||||
}, "question response delivery will retry after wake failure");
|
||||
if (!exhausted) return null;
|
||||
return recordTerminal({
|
||||
delivery: claimed,
|
||||
interaction,
|
||||
status: "failed",
|
||||
mode: null,
|
||||
targetRunId: null,
|
||||
adapter,
|
||||
errorCode,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function sweepPending(limit = 50) {
|
||||
const sweepAt = now();
|
||||
const staleAt = new Date(sweepAt.getTime() - claimStaleMs);
|
||||
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 };
|
||||
}
|
||||
|
|
@ -11,6 +11,7 @@ type RunnerModule = typeof import("@paperclipai/paperclip-runner");
|
|||
|
||||
export type {
|
||||
PaperclipJsonValue,
|
||||
PaperclipQuestionResponse,
|
||||
PaperclipSemanticActionBinding,
|
||||
PaperclipSemanticActionId,
|
||||
PaperclipSemanticAuthorizationRecord,
|
||||
|
|
|
|||
Loading…
Reference in New Issue