feat(runner): integrate native runs into task threads

This commit is contained in:
Dotta committed 2026-08-25 18:03:36 -05:00
1 parent a5c2add7be
commit 10560e64be
28 files changed
+1501 -35

No files matched your search

+2
View File
@@ -5113,6 +5113,7 @@ export function agentRoutes(
const columns = {
id: heartbeatRuns.id,
runtimeMode: heartbeatRuns.runtimeMode,
companyId: heartbeatRuns.companyId,
status: heartbeatRuns.status,
invocationSource: heartbeatRuns.invocationSource,
@@ -5338,6 +5339,7 @@ export function agentRoutes(
const liveRuns = await db
.select({
id: heartbeatRuns.id,
runtimeMode: heartbeatRuns.runtimeMode,
status: heartbeatRuns.status,
invocationSource: heartbeatRuns.invocationSource,
triggerDetail: heartbeatRuns.triggerDetail,
+26 -1
View File
@@ -207,6 +207,11 @@ import { environmentService } from "../services/environments.js";
import { environmentRuntimeService } from "../services/environment-runtime.js";
import { redactSensitiveText } from "../redaction.js";
import { createRunSecretRedactionRegistry } from "../services/run-secret-redaction.js";
import {
deliverNativeQuestionResponse,
nativeQuestionRunToCancel,
validateNativeQuestionResponseInput,
} from "../services/native-runtime/native-question-bridge.js";
import {
createCompanySearchRateLimiter,
type CompanySearchRateLimiter,
@@ -11657,9 +11662,12 @@ export function issueRoutes(
interactionId,
);
if (!authorizedResolution) return;
const { interactionSvc, resolutionAuthorization } = authorizedResolution;
const { interactionSvc, current, resolutionAuthorization } = authorizedResolution;
const actor = getActorInfo(req);
if (current.kind === "ask_user_questions") {
validateNativeQuestionResponseInput(current, req.body);
}
const interaction = await interactionSvc.answerQuestions(issue, interactionId, req.body, {
agentId: actor.agentId,
runId: actor.runId,
@@ -11694,6 +11702,10 @@ export function issueRoutes(
},
});
if (interaction.kind === "ask_user_questions") {
await deliverNativeQuestionResponse(db, interaction);
}
await queueResolvedInteractionContinuationWakeup({
db,
heartbeat,
@@ -11882,6 +11894,19 @@ export function issueRoutes(
},
});
if (interaction.kind === "ask_user_questions") {
const nativeRunId = await nativeQuestionRunToCancel(db, interaction);
if (nativeRunId) {
await heartbeat.cancelRun(nativeRunId, "Cancelled while waiting for operator input", {
resultJson: {
cancelledByActorType: "user",
cancelledByUserId: req.actor.userId ?? null,
cancelledInteractionId: interaction.id,
},
});
}
}
await queueResolvedInteractionContinuationWakeup({
db,
heartbeat,
+1
View File
@@ -381,6 +381,7 @@ export function activityService(db: Db) {
const runs = await db
.select({
runId: heartbeatRuns.id,
runtimeMode: heartbeatRuns.runtimeMode,
status: heartbeatRuns.status,
agentId: heartbeatRuns.agentId,
adapterType: agents.adapterType,
+1
View File
@@ -2496,6 +2496,7 @@ const heartbeatRunLogAccessColumns = {
const heartbeatRunIssueSummaryColumns = {
id: heartbeatRuns.id,
runtimeMode: heartbeatRuns.runtimeMode,
status: heartbeatRuns.status,
invocationSource: heartbeatRuns.invocationSource,
triggerDetail: heartbeatRuns.triggerDetail,
@@ -0,0 +1,319 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import { eq, sql } from "drizzle-orm";
import {
activityLog,
agents,
companies,
createDb,
heartbeatRuns,
issueThreadInteractions,
issues,
} from "@paperclipai/db";
import type { PrpEvent } from "@paperclipai/paperclip-runner";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "../../__tests__/helpers/embedded-postgres.js";
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
import {
deliverNativeQuestionResponse,
flushNativeQuestionResponses,
nativeQuestionBridgeInternals,
nativeQuestionRunToCancel,
projectNativeRuntimeRequest,
registerNativeQuestionCommandTarget,
validateNativeQuestionResponseInput,
} from "./native-question-bridge.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping native question bridge tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
describeEmbeddedPostgres("native question bridge", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
let db: ReturnType<typeof createDb>;
let companyId: string;
let issueId: string;
let agentId: string;
let runId: string;
let sessionId: string;
let runnerInstanceId: string;
beforeAll(async () => {
temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-question-");
db = createDb(temporary.connectionString);
}, 20_000);
afterEach(async () => {
nativeQuestionBridgeInternals.resetForTests();
await db.execute(sql.raw(`
TRUNCATE TABLE
"activity_log",
"issue_thread_interactions",
"heartbeat_runs",
"agent_wakeup_requests",
"issues",
"agents",
"companies"
RESTART IDENTITY CASCADE
`));
});
afterAll(async () => temporary?.cleanup());
async function seed() {
companyId = randomUUID();
issueId = randomUUID();
agentId = randomUUID();
runId = randomUUID();
sessionId = randomUUID();
runnerInstanceId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Native questions",
issuePrefix: `NQ${companyId.replaceAll("-", "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "Native Codex",
adapterType: "paperclip_runner",
status: "running",
adapterConfig: { provider: "codex" },
runtimeConfig: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Answer a native question",
status: "in_progress",
assigneeAgentId: agentId,
});
await db.insert(heartbeatRuns).values({
id: runId,
companyId,
agentId,
status: "running",
runtimeMode: "native",
runtimeModeResolvedAt: new Date(),
nativeIssueId: issueId,
nativeSessionId: sessionId,
runnerInstanceId,
driverKind: "codex",
contextSnapshot: { issueId },
});
}
function runtimeRequestEvent(): PrpEvent {
return {
schema: "paperclip.prp.event.v1",
sourceEventId: "runtime-question-1",
sourceSeq: 1,
sourceInstanceId: runnerInstanceId,
sourceKind: "runner",
runId,
normalizedSessionId: sessionId,
turnId: "turn-1",
itemId: "item-1",
eventType: "runtime_request.created",
schemaVersion: 1,
priority: 0,
emittedAt: "2026-08-25T18:00:00.000Z",
payload: {
request: {
schema: "paperclip.runtime_request.v2",
requestKind: "runtime",
requestId: "request-1",
type: "input",
status: "pending",
prompt: "Choose a deployment color",
input: {
schema: "paperclip.question_set.v1",
title: "Deployment",
questions: [{
id: "color",
prompt: "Which color?",
required: true,
answerMode: "single_select",
options: [
{ id: "blue", label: "Blue" },
{ id: "green", label: "Green" },
],
}],
},
},
},
};
}
function binding() {
return {
companyId,
issueId,
runId,
agentId,
normalizedSessionId: sessionId,
runnerSourceInstanceId: runnerInstanceId,
completionContractId: randomUUID(),
completionContractSha256: `sha256:${"a".repeat(64)}`,
completionContractRevision: "1",
completionContractCriterionIds: [],
};
}
it("materializes, validates, and durably resumes a provider-neutral question response", async () => {
await seed();
const interaction = await projectNativeRuntimeRequest({
db,
binding: binding(),
event: runtimeRequestEvent(),
});
expect(interaction).toMatchObject({
kind: "ask_user_questions",
status: "pending",
sourceRunId: runId,
continuationPolicy: "none",
effectiveResolverPolicy: "human_only",
payload: {
questionSet: { schema: "paperclip.question_set.v1" },
questions: [{
id: "color",
selectionMode: "single",
allowOther: false,
options: [{ id: "blue", label: "Blue" }, { id: "green", label: "Green" }],
}],
},
});
expect(await db.select().from(activityLog)).toHaveLength(1);
const answer = { answers: [{ questionId: "color", optionIds: ["blue"] }] };
validateNativeQuestionResponseInput(interaction!, answer);
expect(() => validateNativeQuestionResponseInput(interaction!, {
answers: [{ questionId: "color", optionIds: ["red"] }],
})).toThrow(/unknown option red/);
const answered = await issueThreadInteractionService(db).answerQuestions(
{ id: issueId, companyId, status: "in_progress" },
interaction!.id,
answer,
{ userId: "operator-1" },
);
const queueCommand = vi.fn(() => ({ commandId: "question", controllerSeq: 1 }));
const release = registerNativeQuestionCommandTarget({
binding: { companyId, issueId, runId, agentId },
queueCommand,
});
await flushNativeQuestionResponses(db, runId);
expect(queueCommand).toHaveBeenCalledWith(
"request.resolve",
{
requestId: "request-1",
response: {
schema: "paperclip.question_response.v1",
answers: { color: { selectedOptionIds: ["blue"] } },
},
},
`question_${interaction!.id}`,
);
expect(answered.kind).toBe("ask_user_questions");
if (answered.kind !== "ask_user_questions") throw new Error("expected question interaction");
await expect(deliverNativeQuestionResponse(db, answered)).resolves.toBe("queued");
await expect(nativeQuestionRunToCancel(db, answered)).resolves.toBe(runId);
release();
});
it("binds projection to the persisted native run and ignores legacy delivery", async () => {
await seed();
const mismatched = runtimeRequestEvent();
mismatched.runId = randomUUID();
await expect(projectNativeRuntimeRequest({ db, binding: binding(), event: mismatched }))
.rejects.toThrow("native_runtime_request_binding_mismatch");
const interaction = await projectNativeRuntimeRequest({
db,
binding: binding(),
event: runtimeRequestEvent(),
});
const answered = await issueThreadInteractionService(db).answerQuestions(
{ id: issueId, companyId, status: "in_progress" },
interaction!.id,
{ answers: [{ questionId: "color", optionIds: ["green"] }] },
{ userId: "operator-1" },
);
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, runId));
expect(answered.kind).toBe("ask_user_questions");
if (answered.kind !== "ask_user_questions") throw new Error("expected question interaction");
await expect(deliverNativeQuestionResponse(db, answered)).resolves.toBe("not_native");
await expect(nativeQuestionRunToCancel(db, answered)).resolves.toBeNull();
});
it("does not duplicate the task card when the runner replays a request", async () => {
await seed();
const first = await projectNativeRuntimeRequest({ db, binding: binding(), event: runtimeRequestEvent() });
const second = await projectNativeRuntimeRequest({ db, binding: binding(), event: runtimeRequestEvent() });
expect(second?.id).toBe(first?.id);
expect(await db.select().from(issueThreadInteractions)).toHaveLength(1);
expect(await db.select().from(activityLog)).toHaveLength(1);
});
it("removes the UI-only marker from a canonical custom response", async () => {
await seed();
const event = runtimeRequestEvent();
const request = event.payload.request as Record<string, unknown>;
const input = request.input as Record<string, unknown>;
input.questions = [{
id: "color",
prompt: "Which color?",
required: true,
answerMode: "single_select",
options: [{ id: "blue", label: "Blue" }],
customAnswer: { enabled: true, label: "Another color" },
}];
const interaction = await projectNativeRuntimeRequest({ db, binding: binding(), event });
const answer = {
answers: [{
questionId: "color",
optionIds: ["paperclip_custom_answer"],
otherText: "purple",
}],
};
validateNativeQuestionResponseInput(interaction!, answer);
const answered = await issueThreadInteractionService(db).answerQuestions(
{ id: issueId, companyId, status: "in_progress" },
interaction!.id,
answer,
{ userId: "operator-1" },
);
expect(answered.kind).toBe("ask_user_questions");
if (answered.kind !== "ask_user_questions") throw new Error("expected question interaction");
const queueCommand = vi.fn(() => ({ commandId: "question", controllerSeq: 1 }));
registerNativeQuestionCommandTarget({
binding: { companyId, issueId, runId, agentId },
queueCommand,
});
await expect(deliverNativeQuestionResponse(db, answered)).resolves.toBe("queued");
expect(queueCommand).toHaveBeenCalledWith(
"request.resolve",
{
requestId: "request-1",
response: {
schema: "paperclip.question_response.v1",
answers: { color: { selectedOptionIds: [], customText: "purple" } },
},
},
`question_${interaction!.id}`,
);
});
});
@@ -0,0 +1,346 @@
import { and, eq } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { heartbeatRuns, issueThreadInteractions } from "@paperclipai/db";
import type {
AskUserQuestionsAnswer,
AskUserQuestionsInteraction,
AskUserQuestionsQuestionOption,
PaperclipQuestionSetPayload,
RespondIssueThreadInteraction,
} from "@paperclipai/shared";
import type { PrpEvent } from "../../vendor/paperclip-runner/index.js";
import {
parsePaperclipQuestionResponse,
parsePaperclipQuestionSet,
type PaperclipQuestionResponse,
type PaperclipQuestionSet,
} from "../../vendor/paperclip-runner/index.js";
import { logger } from "../../middleware/logger.js";
import { unprocessable } from "../../errors.js";
import { logActivity } from "../activity-log.js";
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
import type { NativeRunStoreBinding } from "./native-run-coordinator-store.js";
const QUESTION_KEY_PREFIX = "paperclip-runner-question:";
const REQUEST_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$/;
const TEXT_ANSWER_OPTION_ID = "paperclip_text_answer";
const CUSTOM_ANSWER_OPTION_ID = "paperclip_custom_answer";
type QueueCommand = (
type: string,
payload?: Record<string, unknown>,
commandId?: string,
) => { readonly commandId: string; readonly controllerSeq: number };
interface NativeQuestionCommandTarget {
binding: Pick<NativeRunStoreBinding, "companyId" | "issueId" | "runId" | "agentId">;
queueCommand: QueueCommand;
}
const activeTargets = new Map<string, NativeQuestionCommandTarget>();
function record(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: null;
}
function requestIdForInteraction(
interaction: Pick<AskUserQuestionsInteraction, "idempotencyKey" | "sourceRunId" | "payload">,
): string | null {
if (!interaction.sourceRunId || !interaction.payload.questionSet) return null;
const key = interaction.idempotencyKey;
const expectedPrefix = `${QUESTION_KEY_PREFIX}${interaction.sourceRunId}:`;
if (!key?.startsWith(expectedPrefix)) return null;
const requestId = key.slice(expectedPrefix.length);
return REQUEST_ID_PATTERN.test(requestId) ? requestId : null;
}
function uniqueSyntheticOptionId(existing: readonly string[], preferred: string): string {
const ids = new Set(existing);
if (!ids.has(preferred)) return preferred;
for (let suffix = 2; suffix < 10_000; suffix += 1) {
const candidate = `${preferred}_${suffix}`;
if (!ids.has(candidate)) return candidate;
}
throw new Error("native_question_synthetic_option_exhausted");
}
function toInteractionPayload(questionSet: PaperclipQuestionSet) {
return {
version: 1 as const,
...(questionSet.title ? { title: questionSet.title.slice(0, 240) } : {}),
...(questionSet.submitLabel ? { submitLabel: questionSet.submitLabel.slice(0, 120) } : {}),
questions: questionSet.questions.map((question) => {
const canonicalOptions = question.options ?? [];
const options: AskUserQuestionsQuestionOption[] = canonicalOptions.map((option) => ({
id: option.id,
label: option.label,
...(option.description ? { description: option.description } : {}),
}));
if (question.answerMode === "text") {
options.push({
id: uniqueSyntheticOptionId([], TEXT_ANSWER_OPTION_ID),
label: question.header ?? "Type an answer",
...(question.textValidation?.inputType
? { description: `Expected ${question.textValidation.inputType} input` }
: {}),
freeText: true,
});
} else if (question.customAnswer?.enabled) {
options.push({
id: uniqueSyntheticOptionId(canonicalOptions.map((option) => option.id), CUSTOM_ANSWER_OPTION_ID),
label: question.customAnswer.label ?? "Other",
...(question.customAnswer.placeholder ? { description: question.customAnswer.placeholder } : {}),
freeText: true,
});
}
return {
id: question.id,
prompt: question.prompt,
...((question.helpText || question.header)
? { helpText: question.helpText ?? question.header }
: {}),
selectionMode: question.answerMode === "multi_select" ? "multi" as const : "single" as const,
required: question.required,
allowOther: question.answerMode === "text" || question.customAnswer?.enabled === true,
options,
};
}),
questionSet: questionSet as PaperclipQuestionSetPayload,
};
}
function canonicalResponse(
questionSet: PaperclipQuestionSetPayload,
answers: readonly AskUserQuestionsAnswer[],
): PaperclipQuestionResponse {
const answerByQuestionId = new Map(answers.map((answer) => [answer.questionId, answer]));
const response: PaperclipQuestionResponse = {
schema: "paperclip.question_response.v1",
answers: {},
};
for (const question of questionSet.questions) {
const answer = answerByQuestionId.get(question.id);
if (!answer) continue;
if (question.answerMode === "text") {
response.answers[question.id] = {
...(answer.otherText !== undefined && answer.otherText !== null
? { text: answer.otherText }
: {}),
};
} else {
const customOptionId = question.customAnswer?.enabled
? uniqueSyntheticOptionId(
(question.options ?? []).map((option) => option.id),
CUSTOM_ANSWER_OPTION_ID,
)
: null;
response.answers[question.id] = {
selectedOptionIds: answer.optionIds.filter((optionId) => optionId !== customOptionId),
...(answer.otherText !== undefined && answer.otherText !== null
? { customText: answer.otherText }
: {}),
};
}
}
return parsePaperclipQuestionResponse(questionSet, response);
}
async function authorizedNativeRun(
db: Db,
interaction: Pick<AskUserQuestionsInteraction, "companyId" | "issueId" | "sourceRunId" | "payload" | "idempotencyKey">,
) {
const requestId = requestIdForInteraction(interaction);
if (!requestId || !interaction.sourceRunId) return null;
const run = await db.select({
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
issueId: heartbeatRuns.nativeIssueId,
agentId: heartbeatRuns.agentId,
runtimeMode: heartbeatRuns.runtimeMode,
status: heartbeatRuns.status,
}).from(heartbeatRuns).where(and(
eq(heartbeatRuns.id, interaction.sourceRunId),
eq(heartbeatRuns.companyId, interaction.companyId),
eq(heartbeatRuns.nativeIssueId, interaction.issueId),
eq(heartbeatRuns.runtimeMode, "native"),
)).limit(1).then((rows) => rows[0] ?? null);
return run ? { ...run, requestId } : null;
}
/** Materialize a canonical runtime input request as the existing task-thread card. */
export async function projectNativeRuntimeRequest(input: {
db: Db;
binding: NativeRunStoreBinding;
event: PrpEvent;
}): Promise<AskUserQuestionsInteraction | null> {
if (input.event.eventType !== "runtime_request.created") return null;
if (
input.event.runId !== input.binding.runId
|| input.event.normalizedSessionId !== input.binding.normalizedSessionId
|| input.event.sourceInstanceId !== input.binding.runnerSourceInstanceId
) {
throw new Error("native_runtime_request_binding_mismatch");
}
const request = record(record(input.event.payload)?.request);
if (
!request
|| request.schema !== "paperclip.runtime_request.v2"
|| request.requestKind !== "runtime"
|| request.type !== "input"
|| request.status !== "pending"
|| typeof request.requestId !== "string"
|| !REQUEST_ID_PATTERN.test(request.requestId)
) {
throw new Error("native_runtime_request_invalid");
}
const questionSet = parsePaperclipQuestionSet(request.input);
if (questionSet.questions.some((question) => question.textValidation?.pattern !== undefined)) {
// JavaScript regular expressions have no execution budget. Provider-authored
// patterns therefore stay fail-closed until the runner contract supplies a
// bounded regex dialect rather than exposing the server to catastrophic backtracking.
throw new Error("native_runtime_question_pattern_unsupported");
}
const idempotencyKey = `${QUESTION_KEY_PREFIX}${input.binding.runId}:${request.requestId}`;
const existing = await input.db.select({ id: issueThreadInteractions.id })
.from(issueThreadInteractions)
.where(and(
eq(issueThreadInteractions.companyId, input.binding.companyId),
eq(issueThreadInteractions.issueId, input.binding.issueId),
eq(issueThreadInteractions.idempotencyKey, idempotencyKey),
))
.limit(1)
.then((rows) => rows[0] ?? null);
const interaction = await issueThreadInteractionService(input.db).create(
{ id: input.binding.issueId, companyId: input.binding.companyId },
{
kind: "ask_user_questions",
idempotencyKey,
sourceRunId: input.binding.runId,
resolverPolicy: "human_only",
continuationPolicy: "none",
...(questionSet.title ? { title: questionSet.title.slice(0, 240) } : {}),
...(typeof request.prompt === "string" ? { summary: request.prompt.slice(0, 1000) } : {}),
payload: toInteractionPayload(questionSet),
},
{ agentId: input.binding.agentId, runId: input.binding.runId },
) as AskUserQuestionsInteraction;
if (!existing) {
await logActivity(input.db, {
companyId: input.binding.companyId,
actorType: "agent",
actorId: input.binding.agentId,
agentId: input.binding.agentId,
runId: input.binding.runId,
action: "issue.thread_interaction_created",
entityType: "issue",
entityId: input.binding.issueId,
details: {
interactionId: interaction.id,
interactionKind: interaction.kind,
interactionStatus: interaction.status,
runtimeMode: "native",
},
});
}
if (interaction.status === "answered") {
await deliverNativeQuestionResponse(input.db, interaction);
}
return interaction;
}
/** Validate untrusted board input before the existing interaction service persists it. */
export function validateNativeQuestionResponseInput(
interaction: AskUserQuestionsInteraction,
input: RespondIssueThreadInteraction,
): void {
if (!requestIdForInteraction(interaction) || !interaction.payload.questionSet) return;
try {
canonicalResponse(interaction.payload.questionSet, input.answers);
} catch (error) {
throw unprocessable(
error instanceof Error ? error.message : "Invalid native question response",
{ code: "invalid_question_response" },
);
}
}
/** Queue an answered interaction into the active durable PRP command stream. */
export async function deliverNativeQuestionResponse(
db: Db,
interaction: AskUserQuestionsInteraction,
): Promise<"not_native" | "pending" | "queued"> {
if (interaction.status !== "answered" || !interaction.result || !interaction.payload.questionSet) {
return "not_native";
}
const run = await authorizedNativeRun(db, interaction);
if (!run) return "not_native";
const response = canonicalResponse(interaction.payload.questionSet, interaction.result.answers);
const target = activeTargets.get(run.id);
if (
!target
|| target.binding.companyId !== run.companyId
|| target.binding.issueId !== run.issueId
|| target.binding.agentId !== run.agentId
) {
return "pending";
}
try {
target.queueCommand(
"request.resolve",
{ requestId: run.requestId, response: response as unknown as Record<string, unknown> },
`question_${interaction.id}`,
);
return "queued";
} catch (error) {
logger.warn(
{ err: error, runId: run.id, interactionId: interaction.id },
"native question response remains durable for session recovery",
);
return "pending";
}
}
export async function flushNativeQuestionResponses(
db: Db,
runId: string,
): Promise<void> {
const target = activeTargets.get(runId);
if (!target) return;
const interactions = await issueThreadInteractionService(db).listForIssue(target.binding.issueId);
for (const interaction of interactions) {
if (
interaction.kind === "ask_user_questions"
&& interaction.sourceRunId === runId
&& interaction.status === "answered"
) {
await deliverNativeQuestionResponse(db, interaction);
}
}
}
export function registerNativeQuestionCommandTarget(target: NativeQuestionCommandTarget): () => void {
const existing = activeTargets.get(target.binding.runId);
if (existing) throw new Error("native_question_command_target_conflict");
activeTargets.set(target.binding.runId, target);
return () => {
if (activeTargets.get(target.binding.runId) === target) {
activeTargets.delete(target.binding.runId);
}
};
}
export async function nativeQuestionRunToCancel(
db: Db,
interaction: AskUserQuestionsInteraction,
): Promise<string | null> {
const run = await authorizedNativeRun(db, interaction);
return run && ["queued", "running"].includes(run.status) ? run.id : null;
}
export const nativeQuestionBridgeInternals = {
resetForTests: () => activeTargets.clear(),
};
@@ -13,6 +13,11 @@ import {
import { registerRunnerPrpAuthority } from "../../realtime/runner-prp-ws.js";
import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js";
import {
flushNativeQuestionResponses,
projectNativeRuntimeRequest,
registerNativeQuestionCommandTarget,
} from "./native-question-bridge.js";
import { PaperclipRunnerSemanticAuthority } from "./runner-semantic-authority.js";
const UUID_PATTERN =
@@ -214,7 +219,7 @@ export function runnerPrpCoordinator(
agentId: input.agentId,
});
const semanticTools = await semanticAuthority.listAlwaysAvailableTools();
const nativeStore = new NativeRunCoordinatorStore(db, {
const storeBinding = {
companyId: input.companyId,
issueId: input.issueId,
runId: input.runId,
@@ -225,7 +230,8 @@ export function runnerPrpCoordinator(
completionContractSha256: binding.run.completionContractSha256,
completionContractRevision: String(binding.completionContract.revision),
completionContractCriterionIds: criterionIds,
});
} as const;
const nativeStore = new NativeRunCoordinatorStore(db, storeBinding);
type StoredCompletedRun = NonNullable<Awaited<ReturnType<typeof nativeStore.readCompletedRun>>>;
type CompletedRun = StoredCompletedRun & { readonly providerSessionId?: string };
const withProviderSession = async (stored: StoredCompletedRun): Promise<CompletedRun> => {
@@ -255,6 +261,9 @@ export function runnerPrpCoordinator(
connectionLeaseTtlMs,
onCommittedEvent: async (event) => {
await nativeStore.appendEvent(event);
if (event.eventType === "runtime_request.created") {
await projectNativeRuntimeRequest({ db, binding: storeBinding, event });
}
await nativeStore.reconcileTerminalEvent(event);
if (event.eventType === "run.terminal") {
const stored = await nativeStore.readCompletedRun();
@@ -274,15 +283,38 @@ export function runnerPrpCoordinator(
},
});
const registration = await registerRunnerPrpAuthority({
companyId: input.companyId,
runId: input.runId,
authority,
});
let registration: Awaited<ReturnType<typeof registerRunnerPrpAuthority>>;
try {
registration = await registerRunnerPrpAuthority({
companyId: input.companyId,
runId: input.runId,
authority,
});
} catch (error) {
authority.disconnectActiveRunner();
throw error;
}
let releaseQuestionTarget = () => {};
try {
releaseQuestionTarget = registerNativeQuestionCommandTarget({
binding: storeBinding,
queueCommand: (type, payload = {}, commandId) => {
const command = authority.queueCommand(type, payload, commandId, true);
return { commandId: command.commandId, controllerSeq: command.controllerSeq };
},
});
await flushNativeQuestionResponses(db, input.runId);
} catch (error) {
releaseQuestionTarget();
authority.disconnectActiveRunner();
await registration.release();
throw error;
}
let bootstrapTicket: string;
try {
bootstrapTicket = authority.issueBootstrapTicket(bootstrapTtlMs);
} catch (error) {
releaseQuestionTarget();
authority.disconnectActiveRunner();
await registration.release();
throw error;
@@ -339,6 +371,7 @@ export function runnerPrpCoordinator(
release: async () => {
if (released) return;
released = true;
releaseQuestionTarget();
authority.disconnectActiveRunner();
await registration.release();
},
+5
View File
@@ -18,6 +18,9 @@ export type {
PaperclipSemanticToolCall,
PaperclipSemanticToolDefinition,
PaperclipSemanticToolResult,
PaperclipQuestionResponse,
PaperclipQuestionSet,
PaperclipRuntimeInputRequest,
PrpEvent,
PrpStructuredRunResult,
PrpTerminalState,
@@ -35,6 +38,8 @@ const runner = await import(sourceUrl.href) as RunnerModule;
export const DurablePrpControlPlane = runner.DurablePrpControlPlane;
export const PaperclipSemanticDispatcher = runner.PaperclipSemanticDispatcher;
export const parsePaperclipQuestionSet = runner.parsePaperclipQuestionSet;
export const parsePaperclipQuestionResponse = runner.parsePaperclipQuestionResponse;
export const validatePrpEvent = runner.validatePrpEvent;
export const validatePrpStructuredRunResult =
runner.validatePrpStructuredRunResult;