mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 22:13:59 +01:00
fix(native): preserve governed terminal receipts during observer settlement (#19025)
## Thinking Path > - Paperclip manages work done by AI agents. > - A native turn can yield to a saved human question or approval. > - The provider must stop, and its terminal event must be saved. > - The old drain waits for both the event transaction and its later observers. > - A slow observer can make a saved terminal look unproved. > - This change separates the committed receipt from owned observer work. > - Checkpoint and result acceptance still wait for that work to settle. ## Linked Issues or Issue Description **What happened?** A native governed wait can fail with `governed_wait_provider_terminal_unproven` after its exact interrupted terminal is already in PostgreSQL. The event port waits for postcommit accounting, output and projection callbacks before it returns the receipt. Those callbacks can cross the provider drain deadline. **What did you expect to happen?** A committed terminal proves the event transaction. Its observers remain separate work. A checkpoint or result must not be accepted until those observers settle. A pending human-only question must keep its audience and continuation policy. **Steps to reproduce** Run the real PostgreSQL control-plane regression. Commit a same-turn interrupted terminal. Hold its postcommit observer beyond the existing provider drain. The parent code reports an unproved provider terminal. The fixed code retains the committed receipt and joins the observer before checkpoint or result acceptance. No overlapping repair was found in the current master or open-PR search. Related [#18977](https://github.com/paperclipai/paperclip/pull/18977) covers final conversation-response selection. It does not change this receipt boundary. The source reproduction is consistent with the observed failure. The exact hosted provider code is unavailable, so it does not prove that every observed failure used this branch. ## What Changed - Add an explicit terminal-only option to the internal event port. Ordinary appends keep their existing behavior. - Return that receipt only after the normal transaction and company, agent, run, session, source and revocation checks. - Retain the full observer promise. Join it before checkpoint and result writes. Keep its original rejection and observe it after an aborted join. - Reuse the existing provider drain and checkpoint deadlines. Preserve duplicate and restart replay without another model turn. - Document the receipt and duplicate-observer replay boundaries. Keep human-only and no-continuation cards unchanged. - Assert the actual Codex facade requests deferral only for its interrupted terminal. Isolate empty skill inventory in monitor-only process fixtures while retaining their real scheduler, lifecycle, selection snapshot and three-second deadline. ## Verification - Meaningful parent red: actual native runtime and real PostgreSQL event port. The terminal row commits, but a held observer causes `NativeProviderTerminalFailure` with `governed_wait_provider_terminal_unproven`. - Current candidate `0db24ae74ee42fdf92035d8407fc095e9f73cc6a`: full runtime suite, 203 passed. Real PostgreSQL control-plane, timing and monitor suites, 103 passed. Runner TypeScript and full layered server typechecks passed on the unchanged candidate source tree. The controls cover uncommitted and foreign evidence, explicit false and failed terminals, held and rejecting observers, abort and controller revocation, duplicate receipts, and new-port recovery. The real pending human-only card stays unanswered with continuation policy `none`. Recovery does not call `startTurn` again. - `pnpm -r typecheck`, the production server build (including the real Runner release binary), module boundaries and vendor dependency verification passed on `6d75`; the follow-up changes only tests and documentation. Diff checks pass on the current candidate. The prior `399ad3dc2b` repository-wide `test:run` was stopped as obsolete during the general-server phase after preserving its progress (exit 130, incomplete); its repository build was not reached. Its six completed chat shards passed 1,904 cases, but this is prior-head evidence and not a full-suite pass. Full current-head CI is required before merge. - CI attempt 1 on `6d75` failed one real three-second monitor-settlement assertion; the aggregate verify gate consequently failed. Other jobs passed. The actual CI merge tree was `3037bf6db162da3b0a9172046360401f8449ebc5` (`6d75` into `89703c3`). An unchanged full-file replay reproduced three timeouts. Original-deadline inspection found the legacy process still preparing; local phase timing measured cold skill inventory at 6.629 seconds. The original CI did not retain that phase detail, so its exact subcause remains unproved. The fixture correction supplies empty skill entries for these skill-free process tests and still invokes the selection-snapshot callback. It preserves all assertions and the three-second deadline, with no new drain wait. All 21 cases pass; deliberately restoring the old blocked-status rejection still fails the real blocked-check case with 409. No CI retry was issued. - The prior current-head native review was 4/5 with no production correctness finding. Its requested direct facade-options assertion and duplicate-observer wording correction are included. Current-head CI is green (55 success, 3 skipped), including required verify and E2E; current-head Greptile is 5/5 with no actionable findings. The old native advisory remains tied to `6d75`, and its requested findings are addressed. No new native assessment is claimed or required by protection. - The first draft broke three existing transaction-start timing controls. The no-debt path now keeps the original synchronous start, and all three pass. Two draft type errors were corrected. A further real-PG red found an extra default-append assertion after commit. Removing it restores the original committed receipt after a late abort; the same aborted signal still cannot write a checkpoint or result. Original setup and assertion failures remain in private verification receipts. - Normal merge of current master `3a2249dd5768313f04b2aa3860dfd363492eda16` has no conflicts. The affected production source blobs did not change in that baseline merge. - Local proof uses Node 24.22.2 on Darwin arm64, a real PostgreSQL instance and deterministic provider-boundary fixtures. It makes no model call. No deployed replay or original-task recovery is claimed. ## Risks Deferred observers must remain owned. A missed join could accept incomplete accounting or output. A rejected or stalled observer must prevent acceptance, and an abort must prevent late checkpoint/result writes. The tests exercise these boundaries. The port keeps the existing best-effort trace-publication handling. This change does not turn a PRP terminal into physical-stop or accounting-completion proof. ## Model Used OpenAI `gpt-6.1-sol`, with code and shell tools. The host does not expose the reasoning mode or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
c3715e74a6
commit
22735cfcea
7 files changed
+423
-30
No files matched your search
@@ -1,6 +1,6 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
|
||||
import {
|
||||
costEvents,
|
||||
@@ -32,6 +32,21 @@ import { normalizeIssueExecutionPolicy, parseIssueExecutionState } from "../serv
|
||||
import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.js";
|
||||
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
// These process fixtures do not consume skills. Keep the real scheduler,
|
||||
// execution lifecycle and selection snapshot, without a cold catalog expansion
|
||||
// competing with the monitor's existing three-second settlement assertion.
|
||||
vi.mock("../services/company-skills.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("../services/company-skills.js")>();
|
||||
return { ...actual, companySkillService: (...args: Parameters<typeof actual.companySkillService>) => {
|
||||
const service = actual.companySkillService(...args);
|
||||
return { ...service, withRuntimeSkillEntries: async <T>(_companyId: string,
|
||||
options: Parameters<typeof service.withRuntimeSkillEntries>[1],
|
||||
consume: (entries: Awaited<ReturnType<typeof service.listRuntimeSkillEntries>>) => T | Promise<T>) => {
|
||||
await options.selectVersions?.([]);
|
||||
return consume([]);
|
||||
} };
|
||||
} };
|
||||
});
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_MONITOR_TEST_DATABASE_URL");
|
||||
const embeddedPostgresSupport = external ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
@@ -64,7 +64,7 @@ import { assertCurrentWakeCommentsRead } from "./current-wake-comments.js";
|
||||
import { buildExecutionContinuation } from "../execution-continuation.js";
|
||||
import { childReviewOutcomes } from "./child-review-outcomes.js";
|
||||
import { createPostgresRunDispatchAdapter } from "../../modules/run-dispatch/adapters/postgres.js";
|
||||
import { materializeRuntimeQuestionFallback } from "./native-session-executor.js";
|
||||
import { materializeRuntimeQuestionFallback, nativeGovernedWaitResult } from "./native-session-executor.js";
|
||||
import {
|
||||
subscribeAllCompanyLiveEvents,
|
||||
subscribeCompanyLiveEvents,
|
||||
@@ -314,6 +314,262 @@ describe("PaperclipControlPlanePort conformance", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["settles", "checkpoint timeout"] as const)("retains a durably recorded governed terminal while its postcommit observer is pending: %s", async mode => {
|
||||
const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, issueId: randomUUID(), runId: randomUUID(), sessionId: randomUUID() };
|
||||
const runnerId = randomUUID();
|
||||
const fixtureContractId = randomUUID();
|
||||
await db.insert(issues).values({ id: identity.issueId, companyId: identity.companyId, title: "Human-only waiting checkpoint",
|
||||
status: "in_progress", assigneeAgentId: identity.agentId });
|
||||
const [baseContract] = await db.select().from(completionContracts).where(eq(completionContracts.id, contractId));
|
||||
await db.insert(completionContracts).values({ ...baseContract!, id: fixtureContractId, issueId: identity.issueId });
|
||||
await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId,
|
||||
status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId,
|
||||
runnerInstanceId: runnerId, completionContractId: fixtureContractId, completionContractSha256: contractSha });
|
||||
const interaction = await issueThreadInteractionService(db).create({ id: identity.issueId, companyId: identity.companyId }, {
|
||||
kind: "ask_user_questions", title: "Choose the format", resolverPolicy: "human_only", continuationPolicy: "none",
|
||||
sourceRunId: identity.runId, payload: { version: 1, title: "Choose the format", questions: [{ id: "format",
|
||||
prompt: "Which format?", required: true, selectionMode: "single", options: [{ id: "onsite", label: "On site" }] }] },
|
||||
}, { agentId: identity.agentId, runId: identity.runId });
|
||||
let releaseObserver!: () => void;
|
||||
const observerGate = new Promise<void>(resolve => { releaseObserver = resolve; });
|
||||
let terminalWasRecorded = false;
|
||||
let terminalRecorded!: () => void;
|
||||
const recorded = new Promise<void>(resolve => { terminalRecorded = resolve; });
|
||||
let observerFinished!: () => void;
|
||||
const observerDone = new Promise<void>(resolve => { observerFinished = resolve; });
|
||||
const port = new PaperclipControlPlanePort(db, { ...identity,
|
||||
completionContractId: fixtureContractId, completionContractSha256: contractSha,
|
||||
sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" }, {
|
||||
async onCommittedEvent(event) {
|
||||
if (event.eventType === "turn.interrupted") { terminalWasRecorded = true; terminalRecorded(); try { await observerGate; } finally { observerFinished(); } }
|
||||
},
|
||||
});
|
||||
const contract = { revision: "phase6-v1", objective: "Conformance", criteria: [{ id: "objective", requirement: "Complete conformance" }] };
|
||||
const yielded = nativeGovernedWaitResult({ interaction: { id: interaction.id,
|
||||
title: interaction.title ?? null, summary: interaction.summary ?? null }, completionContract: contract });
|
||||
const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record<string, unknown>): PrpEvent => ({
|
||||
schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:${sourceSeq}`, sourceInstanceId: runnerId, sourceSeq,
|
||||
sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId, turnId: "park-turn",
|
||||
eventType, schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), payload,
|
||||
});
|
||||
let starts = 0, terminalEmitted = false;
|
||||
const backend: NativeSessionBackend = {
|
||||
async descriptor() { return { kind: "mock", name: "park-observer", version: "1",
|
||||
capabilities: { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true } }; },
|
||||
async openSession() { return { identity: () => identity,
|
||||
async capabilities() { return { resume: false, typedEvents: true, steering: false, interruption: true, structuredResult: true }; },
|
||||
async startTurn() { starts++; return { turnId: "park-turn" }; },
|
||||
cancel() { return { cleanup: Promise.resolve() }; },
|
||||
async *events() { yield event(1, "item.completed", { kind: "dynamicToolCall" }); terminalEmitted = true; yield event(2, "turn.interrupted", { status: "interrupted", error: null }); },
|
||||
async snapshot() { return { backendKind: "mock", sessionId: identity.sessionId, identity,
|
||||
activeTurnId: terminalEmitted ? null : "park-turn", cursor: terminalEmitted ? "2" : "1" }; },
|
||||
async result() { return null; }, async close() {},
|
||||
}; },
|
||||
};
|
||||
const options = { backend, controlPlane: port, timeoutMs: 2_000,
|
||||
...(mode === "checkpoint timeout" ? { checkpointTimeoutMs: 100 } : {}),
|
||||
runnerInstanceId: runnerId, controlPlaneInstanceId: "park-control",
|
||||
input: parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v1", binding: {
|
||||
companyId: identity.companyId, runId: identity.runId, issueId: identity.issueId, agentId: identity.agentId, executionWorkspaceId: "park-workspace" },
|
||||
task: { identifier: "PARK-1", title: "Wait for a human", description: null, prompt: "Wait for a human", workMode: "standard" },
|
||||
workspace: { cwd: process.cwd(), repoUrl: null, repoRef: null, branchName: null },
|
||||
session: { normalizedSessionId: identity.sessionId, driverKind: "codex_app_server", protocolVersion: 1 },
|
||||
provider: { kind: "codex", model: null }, interactionResponses: [], credentialBindings: [],
|
||||
completionContract: { id: fixtureContractId, sha256: contractSha, schemaVersion: "paperclip.completion-contract.v1", contract } }),
|
||||
resolveGovernedWait: ({ event }: { event: PrpEvent }) => event.eventType === "item.completed" ? yielded : null,
|
||||
};
|
||||
const execution = executeNativeSession(options);
|
||||
const outcome = execution.then(value => ({ value }), error => ({ error }));
|
||||
try {
|
||||
await Promise.race([recorded, outcome.then(result => { if ("error" in result) throw result.error; throw new Error("run settled before recording terminal"); })]);
|
||||
const rows = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, identity.runId));
|
||||
expect(rows.some(row => row.eventType === "turn.interrupted")).toBe(true);
|
||||
await new Promise(resolve => setTimeout(resolve, 700));
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]);
|
||||
releaseObserver();
|
||||
await observerDone;
|
||||
if (mode === "settles") expect(await outcome).not.toHaveProperty("error");
|
||||
else {
|
||||
expect(await outcome).toMatchObject({ error: { message: "native session checkpoint timed out after 100ms" } });
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]);
|
||||
const restored = new PaperclipControlPlanePort(db, { ...identity, completionContractId: fixtureContractId,
|
||||
completionContractSha256: contractSha, sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "park-control" });
|
||||
const checkpoint = await restored.loadSessionCheckpoint();
|
||||
expect(checkpoint?.governedWait?.sourceEvent).toMatchObject({ sourceSeq: 1, turnId: "park-turn" });
|
||||
const session = await backend.openSession({ identity });
|
||||
const recovering = { ...backend, async recoverSession() { return { recovered: true as const, session }; } };
|
||||
await expect(executeNativeSession({ ...options, checkpointTimeoutMs: 1_000, controlPlane: restored,
|
||||
backend: recovering, persistedSession: checkpoint })).resolves.toMatchObject({ result: yielded });
|
||||
}
|
||||
expect(starts).toBe(1);
|
||||
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0]).toMatchObject({
|
||||
status: "pending", result: null, effectiveResolverPolicy: "human_only", continuationPolicy: "none", sourceRunId: identity.runId,
|
||||
});
|
||||
} finally { releaseObserver(); await outcome; if (terminalWasRecorded) await observerDone; }
|
||||
});
|
||||
|
||||
async function terminalObserverFixture(options: {
|
||||
onCommittedEvent?: (event: PrpEvent) => Promise<void>;
|
||||
onDuplicateEvent?: (event: PrpEvent) => Promise<void>;
|
||||
assertControllerActive?: () => void;
|
||||
} = {}) {
|
||||
const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
|
||||
const runnerId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: identity.runId, companyId: identity.companyId, agentId: identity.agentId,
|
||||
status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, nativeSessionId: identity.sessionId,
|
||||
runnerInstanceId: runnerId, completionContractId: contractId, completionContractSha256: contractSha });
|
||||
const binding = { ...identity, completionContractId: contractId, completionContractSha256: contractSha,
|
||||
sourceInstanceId: runnerId, controlPlaneSourceInstanceId: "observer-control" };
|
||||
const port = new PaperclipControlPlanePort(db, binding, options);
|
||||
await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
|
||||
const event: PrpEvent = { schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceInstanceId: runnerId,
|
||||
sourceSeq: 1, sourceKind: "runner", runId: identity.runId, normalizedSessionId: identity.sessionId,
|
||||
turnId: "observer-turn", eventType: "turn.interrupted", schemaVersion: 1, priority: 0,
|
||||
emittedAt: new Date().toISOString(), payload: { status: "interrupted", error: null } };
|
||||
const snapshot = { backendKind: "mock" as const, sessionId: identity.sessionId, identity };
|
||||
const result = { result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL };
|
||||
return { identity, runnerId, binding, port, event, snapshot, result };
|
||||
}
|
||||
|
||||
it.each(["default", "deferred"] as const)("keeps %s terminal observation owned before checkpoint and result acceptance", async mode => {
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
let observed = 0;
|
||||
const f = await terminalObserverFixture({ async onCommittedEvent() { observed++; await gate; } });
|
||||
const signal = new AbortController().signal;
|
||||
let appendSettled = false;
|
||||
const append = f.port.appendEvent(f.event, { signal, ...(mode === "deferred" ? { deferPostCommitObservers: true } : {}) })
|
||||
.then(receipt => { appendSettled = true; return receipt; });
|
||||
let checkpointSettled = false, resultSettled = false;
|
||||
try {
|
||||
if (mode === "deferred") {
|
||||
expect(await append).toMatchObject({ disposition: "committed" });
|
||||
const checkpoint = f.port.checkpointSession(f.snapshot, { signal }).then(() => { checkpointSettled = true; });
|
||||
const result = f.port.completeRun(f.result, { signal }).then(() => { resultSettled = true; });
|
||||
await new Promise(resolve => setTimeout(resolve, 30));
|
||||
expect(checkpointSettled).toBe(false);
|
||||
expect(resultSettled).toBe(false);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined();
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]);
|
||||
release(); await Promise.all([checkpoint, result]);
|
||||
} else {
|
||||
await new Promise(resolve => setTimeout(resolve, 30));
|
||||
expect(appendSettled).toBe(false);
|
||||
}
|
||||
} finally { release(); await append; }
|
||||
expect(observed).toBe(1);
|
||||
});
|
||||
|
||||
it("returns the default committed receipt when cancellation arrives during its postcommit observer", async () => {
|
||||
let release!: () => void, observed!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const started = new Promise<void>(resolve => { observed = resolve; });
|
||||
const f = await terminalObserverFixture({ async onCommittedEvent() { observed(); await gate; } });
|
||||
const controller = new AbortController();
|
||||
const outcome = f.port.appendEvent(f.event, { signal: controller.signal }).then(receipt => ({ receipt }), error => ({ error }));
|
||||
try {
|
||||
await started;
|
||||
expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.identity.runId))).toHaveLength(1);
|
||||
const cancellation = new Error("cancelled after event transaction committed");
|
||||
controller.abort(cancellation);
|
||||
release();
|
||||
expect(await outcome).toMatchObject({ receipt: { disposition: "committed" } });
|
||||
await expect(f.port.checkpointSession(f.snapshot, { signal: controller.signal })).rejects.toBe(cancellation);
|
||||
await expect(f.port.completeRun(f.result, { signal: controller.signal })).rejects.toBe(cancellation);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]);
|
||||
} finally { release(); await outcome; }
|
||||
});
|
||||
|
||||
it("retains the original deferred observer rejection for checkpoint and result acceptance", async () => {
|
||||
let reject!: (error: Error) => void;
|
||||
const gate = new Promise<void>((_resolve, no) => { reject = no; });
|
||||
const original = new Error("accounting persistence rejected");
|
||||
const f = await terminalObserverFixture({ async onCommittedEvent() { await gate; } });
|
||||
const signal = new AbortController().signal;
|
||||
await f.port.appendEvent(f.event, { signal, deferPostCommitObservers: true });
|
||||
const checkpoint = f.port.checkpointSession(f.snapshot, { signal }).catch(error => error);
|
||||
const result = f.port.completeRun(f.result, { signal }).catch(error => error);
|
||||
reject(original);
|
||||
expect(await checkpoint).toBe(original); expect(await result).toBe(original);
|
||||
await expect(f.port.completeRun(f.result)).rejects.toBe(original);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined();
|
||||
});
|
||||
|
||||
it.each(["abort", "detach"] as const)("revokes a checkpoint and result waiting for deferred observation on %s", async reason => {
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
let detached = false;
|
||||
const original = new Error("controller authority revoked");
|
||||
const f = await terminalObserverFixture({ async onCommittedEvent() { await gate; },
|
||||
assertControllerActive() { if (detached) throw original; } });
|
||||
const controller = new AbortController();
|
||||
await f.port.appendEvent(f.event, { signal: controller.signal, deferPostCommitObservers: true });
|
||||
const checkpoint = f.port.checkpointSession(f.snapshot, { signal: controller.signal }).catch(error => error);
|
||||
const result = f.port.completeRun(f.result, { signal: controller.signal }).catch(error => error);
|
||||
if (reason === "abort") controller.abort(original); else detached = true;
|
||||
release();
|
||||
expect(await checkpoint).toBe(original); expect(await result).toBe(original);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined();
|
||||
});
|
||||
|
||||
it("keeps a later observer error visible after an aborted checkpoint stops waiting", async () => {
|
||||
let reject!: (error: Error) => void;
|
||||
const gate = new Promise<void>((_resolve, no) => { reject = no; });
|
||||
const f = await terminalObserverFixture({ async onCommittedEvent() { await gate; } });
|
||||
const controller = new AbortController();
|
||||
await f.port.appendEvent(f.event, { signal: controller.signal, deferPostCommitObservers: true });
|
||||
const checkpoint = f.port.checkpointSession(f.snapshot, { signal: controller.signal }).catch(error => error);
|
||||
const stop = new Error("checkpoint authority revoked");
|
||||
controller.abort(stop);
|
||||
expect(await checkpoint).toBe(stop);
|
||||
const original = new Error("retained accounting failure after revocation");
|
||||
reject(original);
|
||||
await expect(f.port.completeRun(f.result)).rejects.toBe(original);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.identity.runId)))[0]!.runnerProfileJson?.sessionCheckpoint).toBeUndefined();
|
||||
});
|
||||
|
||||
it("joins the exact deferred observer on duplicate append and replays it after a new port opens", async () => {
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
let observed = 0, replayed = 0;
|
||||
const f = await terminalObserverFixture({ async onCommittedEvent() { observed++; await gate; },
|
||||
async onDuplicateEvent() { throw new Error("must join the original owned observation"); } });
|
||||
const signal = new AbortController().signal;
|
||||
await f.port.appendEvent(f.event, { signal, deferPostCommitObservers: true });
|
||||
let duplicateSettled = false;
|
||||
const duplicate = f.port.appendEvent(f.event, { signal }).then(value => { duplicateSettled = true; return value; });
|
||||
try {
|
||||
await new Promise(resolve => setTimeout(resolve, 30)); expect(duplicateSettled).toBe(false);
|
||||
release(); expect(await duplicate).toMatchObject({ disposition: "duplicate" });
|
||||
expect(observed).toBe(1);
|
||||
const restored = new PaperclipControlPlanePort(db, f.binding, { async onDuplicateEvent() { replayed++; } });
|
||||
await restored.openRun({ identity: f.identity, backendKind: "mock", sourceInstanceId: f.runnerId });
|
||||
await expect(restored.appendEvent(f.event)).resolves.toMatchObject({ disposition: "duplicate" });
|
||||
expect(replayed).toBe(1);
|
||||
expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.identity.runId))).toHaveLength(1);
|
||||
} finally { release(); await duplicate; }
|
||||
});
|
||||
|
||||
it.each(["company", "agent", "run", "session", "source", "failed", "unproven", "nonterminal"] as const)("does not grant deferred terminal authority to %s evidence", async fault => {
|
||||
const f = await terminalObserverFixture();
|
||||
const event = { ...f.event,
|
||||
...(fault === "run" ? { runId: randomUUID() } : {}),
|
||||
...(fault === "session" ? { normalizedSessionId: randomUUID() } : {}),
|
||||
...(fault === "source" ? { sourceInstanceId: randomUUID() } : {}),
|
||||
...(fault === "failed" ? { eventType: "turn.failed" as const } : {}),
|
||||
...(fault === "unproven" ? { payload: { providerTerminalObserved: false } } : {}),
|
||||
...(fault === "nonterminal" ? { eventType: "item.completed" as const } : {}),
|
||||
};
|
||||
const port = fault === "company" || fault === "agent" ? new PaperclipControlPlanePort(db, { ...f.binding,
|
||||
...(fault === "company" ? { companyId: randomUUID() } : { agentId: randomUUID() }) }) : f.port;
|
||||
await expect(port.appendEvent(event, { signal: new AbortController().signal, deferPostCommitObservers: true })).rejects.toThrow();
|
||||
expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.identity.runId))).toEqual([]);
|
||||
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.identity.runId))).toEqual([]);
|
||||
});
|
||||
|
||||
it("publishes native recovery metadata in the existing snapshot transaction and fences stale generations", async () => {
|
||||
const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
|
||||
const runnerId = randomUUID();
|
||||
@@ -406,6 +662,7 @@ describe("PaperclipControlPlanePort conformance", () => {
|
||||
it.each([
|
||||
["checkpoint", "detach"], ["result", "detach"], ["event", "detach"],
|
||||
["checkpoint", "abort"], ["result", "abort"], ["event", "abort"],
|
||||
["deferred event", "detach"], ["deferred event", "abort"],
|
||||
] as const)("revokes an in-flight %s on %s", async (operation, reason) => {
|
||||
const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() };
|
||||
const runnerId = randomUUID();
|
||||
@@ -424,12 +681,13 @@ describe("PaperclipControlPlanePort conformance", () => {
|
||||
assertControllerActive: () => { if (detached) throw new Error("controller_detached"); },
|
||||
});
|
||||
await port.openRun({ identity, backendKind: "mock", sourceInstanceId: runnerId });
|
||||
const mutate = () => operation === "event" ? port.appendEvent({
|
||||
const mutate = () => operation === "event" || operation === "deferred event" ? port.appendEvent({
|
||||
schema: "paperclip.prp.event.v1", sourceEventId: `${runnerId}:1`, sourceSeq: 1,
|
||||
sourceInstanceId: runnerId, sourceKind: "runner", runId: identity.runId,
|
||||
normalizedSessionId: identity.sessionId, eventType: "turn.interrupted", schemaVersion: 1,
|
||||
turnId: "revoked-turn",
|
||||
priority: 0, emittedAt: new Date().toISOString(), payload: { reason: "governed_wait" },
|
||||
}, { signal: controller.signal }) : operation === "checkpoint"
|
||||
}, { signal: controller.signal, ...(operation === "deferred event" ? { deferPostCommitObservers: true } : {}) }) : operation === "checkpoint"
|
||||
? port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity }, { signal: controller.signal })
|
||||
: port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }, { signal: controller.signal });
|
||||
let pending: Promise<unknown> | undefined;
|
||||
|
||||
@@ -98,6 +98,8 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
readonly #nativeRecovery?: { compatibility: NativeControllerCompatibility; owner: NativeRunnerCheckpointOwner };
|
||||
readonly #taskSessionPublication?: NativeTaskSessionPublicationContext;
|
||||
readonly #taskSessionPublicationOwner?: NativeRunnerCheckpointOwner;
|
||||
readonly #deferredEventObservers = new Map<string, Promise<void>>();
|
||||
#deferredObserverFailure: { error: unknown } | null = null;
|
||||
|
||||
constructor(
|
||||
db: Db,
|
||||
@@ -137,6 +139,33 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
options?.signal.throwIfAborted();
|
||||
}
|
||||
|
||||
async #settleDeferredObservers(options?: { signal: AbortSignal }): Promise<void> {
|
||||
const assertActive = () => {
|
||||
this.#assertActive(options);
|
||||
if (this.#deferredObserverFailure) throw this.#deferredObserverFailure.error;
|
||||
};
|
||||
assertActive();
|
||||
if (this.#deferredEventObservers.size === 0) return;
|
||||
const signal = options?.signal;
|
||||
let removeAbort = () => {};
|
||||
const aborted = new Promise<never>((_resolve, reject) => {
|
||||
if (!signal) return;
|
||||
const abort = () => reject(signal.reason ?? new Error("native event observation aborted"));
|
||||
signal.addEventListener("abort", abort, { once: true });
|
||||
removeAbort = () => signal.removeEventListener("abort", abort);
|
||||
if (signal.aborted) abort();
|
||||
});
|
||||
try {
|
||||
// Aborting a checkpoint revokes its writes, not already-owned accounting
|
||||
// or output observation. Each retained promise remains observed below.
|
||||
while (this.#deferredEventObservers.size > 0) {
|
||||
await Promise.race([Promise.all([...this.#deferredEventObservers.values()]), aborted]);
|
||||
assertActive();
|
||||
}
|
||||
} finally { removeAbort(); }
|
||||
assertActive();
|
||||
}
|
||||
|
||||
#matchesPersistedBinding(run: Pick<typeof heartbeatRuns.$inferSelect,
|
||||
"companyId" | "agentId" | "runtimeMode" | "nativeIssueId" | "nativeSessionId" |
|
||||
"runnerInstanceId" | "completionContractId" | "completionContractSha256">): boolean {
|
||||
@@ -207,6 +236,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
|
||||
async checkpointSession(snapshot: PersistedNativeSession, options?: { signal: AbortSignal }): Promise<void> {
|
||||
this.#assertActive(options);
|
||||
if (this.#deferredEventObservers.size > 0 || this.#deferredObserverFailure) await this.#settleDeferredObservers(options);
|
||||
snapshot = redactRunnerApiPayloads(this.#identityRedactor.redact(snapshot));
|
||||
const identity = snapshot.identity;
|
||||
if (
|
||||
@@ -324,7 +354,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
});
|
||||
}
|
||||
|
||||
async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal }) {
|
||||
async appendEvent(value: NativeRunEvent | PrpEvent, options?: { signal: AbortSignal; deferPostCommitObservers?: true }) {
|
||||
const persistenceStartedAt = Date.now(), persistenceClock = performance.now();
|
||||
this.#assertActive(options);
|
||||
if (!isPrpEvent(value)) throw new Error("native_legacy_event_not_supported");
|
||||
@@ -340,6 +370,10 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
if (event.sourceInstanceId !== expectedSourceInstanceId) {
|
||||
throw new Error("native_event_source_binding_mismatch");
|
||||
}
|
||||
if (options?.deferPostCommitObservers && (
|
||||
!["turn.completed", "turn.interrupted", "turn.cancelled"].includes(event.eventType)
|
||||
|| !event.turnId || event.payload.providerTerminalObserved === false
|
||||
)) throw new Error("native_deferred_observer_terminal_required");
|
||||
const terminalItem = /^item\.(completed|failed|cancelled)$/.test(event.eventType);
|
||||
const terminalTurn = /^(turn|session)\.(completed|failed|cancelled|interrupted|closed)$/.test(event.eventType);
|
||||
if ((event.eventType === "item.delta" || terminalItem || terminalTurn) && this.#identityRedactor.values.length > 0) {
|
||||
@@ -396,32 +430,48 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
return receipt;
|
||||
});
|
||||
this.#assertActive(options);
|
||||
try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceStartedAt + performance.now() - persistenceClock); } catch {}
|
||||
if (persisted.disposition === "committed") {
|
||||
try {
|
||||
publishLiveEvent({
|
||||
companyId: this.#binding.companyId,
|
||||
type: "heartbeat.run.event",
|
||||
payload: {
|
||||
runId: this.#binding.runId,
|
||||
agentId: this.#binding.agentId,
|
||||
issueId: this.#binding.issueId,
|
||||
seq: persisted.row.seq,
|
||||
eventType: persisted.row.eventType,
|
||||
stream: persisted.row.stream,
|
||||
level: persisted.row.level,
|
||||
payload: persisted.row.payload,
|
||||
},
|
||||
});
|
||||
} catch {
|
||||
const persistenceEndedAt = persistenceStartedAt + performance.now() - persistenceClock;
|
||||
const observerKey = JSON.stringify([event.sourceInstanceId, event.sourceEventId]);
|
||||
let observation = this.#deferredEventObservers.get(observerKey);
|
||||
if (!observation) observation = (async () => {
|
||||
try { await this.#onEventPersistence?.(event, persistenceStartedAt, persistenceEndedAt); } catch {}
|
||||
if (persisted.disposition === "committed") {
|
||||
try {
|
||||
publishLiveEvent({
|
||||
companyId: this.#binding.companyId,
|
||||
type: "heartbeat.run.event",
|
||||
payload: {
|
||||
runId: this.#binding.runId,
|
||||
agentId: this.#binding.agentId,
|
||||
issueId: this.#binding.issueId,
|
||||
seq: persisted.row.seq,
|
||||
eventType: persisted.row.eventType,
|
||||
stream: persisted.row.stream,
|
||||
level: persisted.row.level,
|
||||
payload: persisted.row.payload,
|
||||
},
|
||||
});
|
||||
} catch {
|
||||
}
|
||||
await this.#onCommittedEvent?.(event, persisted.row.seq);
|
||||
} else {
|
||||
// A recovered runner may replay the event whose durable side effects
|
||||
// parked the prior attempt. Do not repeat those effects, but let the
|
||||
// embedding runtime refresh observational state before appendEvent
|
||||
// returns to its synchronous governed-wait boundary.
|
||||
await this.#onDuplicateEvent?.(event, persisted.row.seq);
|
||||
}
|
||||
await this.#onCommittedEvent?.(event, persisted.row.seq);
|
||||
})();
|
||||
if (options?.deferPostCommitObservers) {
|
||||
this.#deferredEventObservers.set(observerKey, observation);
|
||||
void observation.then(() => {
|
||||
if (this.#deferredEventObservers.get(observerKey) === observation) this.#deferredEventObservers.delete(observerKey);
|
||||
}, error => {
|
||||
this.#deferredObserverFailure ??= { error };
|
||||
if (this.#deferredEventObservers.get(observerKey) === observation) this.#deferredEventObservers.delete(observerKey);
|
||||
});
|
||||
} else {
|
||||
// A recovered runner may replay the event whose durable side effects
|
||||
// parked the prior attempt. Do not repeat those effects, but let the
|
||||
// embedding runtime refresh observational state before appendEvent
|
||||
// returns to its synchronous governed-wait boundary.
|
||||
await this.#onDuplicateEvent?.(event, persisted.row.seq);
|
||||
await observation;
|
||||
}
|
||||
return {
|
||||
cursor: persisted.row.seq,
|
||||
@@ -462,6 +512,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort {
|
||||
|
||||
async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise<void> {
|
||||
this.#assertActive(options);
|
||||
if (this.#deferredEventObservers.size > 0 || this.#deferredObserverFailure) await this.#settleDeferredObservers(options);
|
||||
value = redactRunnerApiPayloads(this.#identityRedactor.redact(value));
|
||||
if (!isCompleteInput(value)) throw new Error("native_structured_result_required");
|
||||
// Capture compatibility diagnostics before canonical validation removes
|
||||
|
||||
Reference in new issue
Block a user