mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 11:01:16 +01:00
fix(native-runner): complete recovery and conformance
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
f285bbbab8
commit
386433f178
22 files changed
+1454
-223
No files matched your search
@@ -703,16 +703,30 @@ from the provisional list above. They are added without expanding authority:
|
||||
```text
|
||||
packages/paperclip-runner/src/native-session-runtime.ts
|
||||
packages/paperclip-runner/src/native-session-runtime.test.ts
|
||||
packages/paperclip-runner/src/contracts/native-execution.ts
|
||||
packages/paperclip-runner/src/contracts/native-execution.test.ts
|
||||
packages/paperclip-runner/src/backends/codex-native-backend.ts
|
||||
packages/paperclip-runner/spec/fixtures/status-authority-phase5.json
|
||||
packages/adapter-utils/package.json
|
||||
packages/shared/src/index.ts
|
||||
packages/shared/src/types/index.ts
|
||||
packages/shared/src/validators/index.ts
|
||||
server/src/services/native-runtime/native-session-executor.ts
|
||||
server/src/services/native-runtime/native-session-executor.test.ts
|
||||
server/src/services/native-runtime/evidence-classifier.test.ts
|
||||
server/src/services/native-runtime/paperclip-control-plane-port.test.ts
|
||||
server/src/services/native-runtime/status-arbiter.test.ts
|
||||
server/src/services/recovery/service.ts
|
||||
packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md
|
||||
```
|
||||
|
||||
The package runtime files own persisted provider-session recovery. The corpus
|
||||
The package runtime, closed native-input contract, and Codex backend files own
|
||||
persisted provider-session recovery and construction. The adapter-utils
|
||||
metadata declares the package dependency used by the approved native
|
||||
finalization result type; the shared barrels only expose the already-reviewed
|
||||
types and validators. The recovery service touch reuses the existing
|
||||
source-scoped recovery action and wake path rather than creating new authority.
|
||||
The corpus
|
||||
edit moves two existing coverage labels between fixtures without changing any
|
||||
fixture input or expected outcome, eliminating an unjoinable fixture. The server
|
||||
executor files own only the run-scoped cancellation handle and coordinator
|
||||
|
||||
@@ -40,8 +40,9 @@ contract are persisted before provider execution.
|
||||
- `completeRun` validates and idempotently persists the immutable structured
|
||||
result plus terminal fact.
|
||||
|
||||
The package-owned `HarnessDriverBackend` is the approved
|
||||
`NativeSessionBackend`. Core passes only the closed `NativeExecutionInputV1`.
|
||||
The package-owned Codex native backend is the production
|
||||
`NativeSessionBackend`; the harness backend remains the deterministic test
|
||||
adapter. Core passes only the closed `NativeExecutionInputV1`.
|
||||
The model receives the smaller `NativeModelEnvelopeV1`; bindings and credential
|
||||
references do not cross that boundary.
|
||||
|
||||
@@ -64,14 +65,23 @@ status-version CAS and audit record.
|
||||
|
||||
## Recovery and safety
|
||||
|
||||
The finalization reconciler selects persisted `runtime_mode=native` rows; it
|
||||
does not consult the current flag. Per-attempt leases prevent concurrent
|
||||
session/finalization ownership, while retryable failures retain the immutable
|
||||
result and a first-class recovery action. A recovered session replays its
|
||||
checkpoint and appends only missing control-plane facts instead of opening a
|
||||
second provider session. Cancellation uses a run-scoped normalized session
|
||||
handle before the existing process cleanup. Native execution does not construct
|
||||
a Paperclip JWT, managed MCP access, legacy context, or raw adapter environment.
|
||||
Recovery selects persisted `runtime_mode=native` coordinators and does not
|
||||
consult the current flag. Result-less transport loss retains the original run,
|
||||
closed native input, provider checkpoint, retry attempt, and recovery action;
|
||||
a database lease dispatches that same run after restart. Missing checkpoints,
|
||||
cancellation, and retry exhaustion fail closed without opening a new provider
|
||||
session or falling into legacy. Result-bearing rows continue through the
|
||||
workspace/finalization reconciler. A recovered package session inspects the
|
||||
provider snapshot before starting a turn, so an active provider turn is not
|
||||
duplicated. Cancellation uses a run-scoped normalized session handle before the
|
||||
existing process cleanup. Native execution does not construct a Paperclip JWT,
|
||||
managed MCP access, legacy context, or raw adapter environment.
|
||||
|
||||
Resolved `request_confirmation` and `ask_user_questions` wakes are projected
|
||||
through the existing authorized issue-interaction service and bound to the
|
||||
company, issue, run, agent, and interaction identities in the persisted input.
|
||||
Governed tool actions, unsupported kinds, unresolved rows, and self-resolution
|
||||
fail closed; no credentials enter the model envelope.
|
||||
|
||||
## Verification status
|
||||
|
||||
|
||||
@@ -51,21 +51,26 @@ pnpm --filter @paperclipai/server exec vitest run \
|
||||
src/__tests__/native-runner-input-boundary.test.ts \
|
||||
src/__tests__/native-run-finalizer.test.ts \
|
||||
src/__tests__/native-status-arbiter-corpus.test.ts \
|
||||
src/__tests__/native-interaction-bridge.test.ts \
|
||||
src/__tests__/native-session-resumption.test.ts \
|
||||
src/__tests__/native-finalization-recovery.test.ts \
|
||||
src/__tests__/native-finalization-migration.test.ts \
|
||||
src/__tests__/legacy-finalization-regression.test.ts
|
||||
src/__tests__/legacy-finalization-regression.test.ts \
|
||||
src/services/native-runtime/native-session-executor.test.ts
|
||||
```
|
||||
|
||||
This runs the unchanged package conformance suite against the real Paperclip
|
||||
port. Twenty-eight targeted tests check company binding, duplicate/gap replay,
|
||||
port. Thirty-nine targeted tests check company binding, duplicate/gap replay,
|
||||
checkpoint recovery, immutable result ingestion, durable-evidence authority,
|
||||
workspace-gated finalization, atomic liveness rollback, status-version CAS,
|
||||
migration repair, legacy byte equivalence, and audit output. All database tests
|
||||
migration repair, legacy byte equivalence, typed interaction materialization,
|
||||
same-run pre-result resumption, lease races, and audit output. All database tests
|
||||
start embedded PostgreSQL and do not skip when a developer database URL is
|
||||
absent.
|
||||
|
||||
The command above is the repeatable internal canary and post-kill-switch legacy
|
||||
proof. The remaining sections are an optional live-provider rollout procedure;
|
||||
The command above is the repeatable scripted internal canary and actual
|
||||
flag-off legacy-heartbeat proof. It is not a live-provider canary. The remaining
|
||||
sections are an optional live-provider rollout procedure;
|
||||
they were not run during remediation and require explicit operator authority to
|
||||
change an instance flag and agent profile.
|
||||
|
||||
|
||||
@@ -30,7 +30,7 @@ pnpm --filter @paperclipai/paperclip-runner trace:phase6 -- \
|
||||
--target mock --scenario happy-path
|
||||
```
|
||||
|
||||
Result: four files and six tests passed. The stable trace reported native
|
||||
Result: four files and seven tests passed. The stable trace reported native
|
||||
mode, `runStatus=succeeded`, three events, contiguous source sequence 3,
|
||||
`workspaceFinalizeStatus=succeeded`, zero legacy invocations, and a server-owned
|
||||
`done` decision with status version 0 -> 1. Restart recovery reused the bound
|
||||
@@ -39,7 +39,7 @@ existing accepted result event, and appended only the missing terminal fact.
|
||||
|
||||
# Real Paperclip adapter and public-session task
|
||||
|
||||
The complete acceptance gate below passed 28 tests in ten files against
|
||||
The complete acceptance gate below passed 39 tests in thirteen files against
|
||||
embedded PostgreSQL with zero skipped tests. Database-backed tests start their
|
||||
own database instead of skipping when a developer `DATABASE_URL` is absent.
|
||||
The same package conformance suite passed unchanged against
|
||||
@@ -59,7 +59,14 @@ The focused corpus also proved:
|
||||
- remote, non-standard, or non-Codex explicit profiles fail closed;
|
||||
- company/run/agent/issue/source bindings reject mismatches;
|
||||
- source gaps remain visible, identical retries deduplicate, and replay is
|
||||
exclusive and byte-stable;
|
||||
exclusive and byte-stable; mutated identities, sequences, results, and
|
||||
bindings fail closed in the shared mock/real port suite;
|
||||
- result-less transport loss dispatches the original persisted native run from
|
||||
its closed envelope/provider checkpoint after a database lease, including
|
||||
flag-off restart and active-turn recovery without a second session or turn;
|
||||
- resolved confirmations and question answers enter the persisted native input
|
||||
only through the authorized issue-interaction service; governed, unresolved,
|
||||
unsupported, or self-approved paths fail closed without credentials;
|
||||
- incomplete/review/yield/cancelled results receive a durable review,
|
||||
continuation, or recovery path;
|
||||
- only a task-wide blocker with a named owner and action selects `blocked`;
|
||||
@@ -72,7 +79,7 @@ The focused corpus also proved:
|
||||
bindings, recovery, status,
|
||||
decisions, and effects roll back together before retry ownership is recorded.
|
||||
|
||||
The acceptance-matrix entry points also passed as one 10-file, 28-test gate:
|
||||
The acceptance-matrix entry points passed as one 13-file, 39-test gate:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/server exec vitest run \
|
||||
@@ -82,16 +89,22 @@ pnpm --filter @paperclipai/server exec vitest run \
|
||||
src/__tests__/native-runner-input-boundary.test.ts \
|
||||
src/__tests__/native-runner-phase6.integration.test.ts \
|
||||
src/__tests__/native-finalization-recovery.test.ts \
|
||||
src/__tests__/native-interaction-bridge.test.ts \
|
||||
src/__tests__/native-session-resumption.test.ts \
|
||||
src/__tests__/heartbeat-run-event-sequencing.test.ts \
|
||||
src/__tests__/legacy-finalization-regression.test.ts \
|
||||
src/__tests__/native-finalization-migration.test.ts \
|
||||
src/__tests__/native-status-arbiter-corpus.test.ts
|
||||
src/__tests__/native-status-arbiter-corpus.test.ts \
|
||||
src/services/native-runtime/native-session-executor.test.ts
|
||||
```
|
||||
|
||||
This gate proves concurrent event allocation, duplicate-only migration repair,
|
||||
bounded retry exhaustion, kill-switch legacy byte compatibility, and complete
|
||||
Section 18.13 fixture-to-consumer traceability in addition to the direct suites.
|
||||
It is the repeatable internal native canary and post-kill-switch legacy proof.
|
||||
bounded retry exhaustion, actual flag-off legacy execution, and executable
|
||||
Section 18.13 fixture-to-consumer traceability. Each of the 52 fixtures and 70
|
||||
unique matrix rows invokes production consumers; the test hashes observed
|
||||
decisions/effects, never the fixture's `expected` object. The selected-task
|
||||
canary remains scripted at the provider boundary, while the legacy heartbeat is
|
||||
an actual adapter/finalizer execution.
|
||||
|
||||
# Compile and migration checks
|
||||
|
||||
@@ -118,14 +131,17 @@ pnpm --filter @paperclipai/server exec vitest run \
|
||||
src/__tests__/heartbeat-run-event-sequencing.test.ts
|
||||
```
|
||||
|
||||
Result: five files and 11 tests passed with zero skips. The production-shaped
|
||||
0211 rehearsal began with legacy event sequences `[1,5,5,9]`, retained the
|
||||
Result: five files and 11 tests passed with zero skips. The 0211 rehearsal
|
||||
reconstructed a production-shaped pre-0211 schema, removed the 0211 journal
|
||||
entry, seeded legacy event sequences `[1,5,5,9]`, and applied the complete
|
||||
migration. It retained the
|
||||
original unique cursors and first `5`, moved only the later duplicate to `10`,
|
||||
seeded `next_event_seq=11`, preserved every non-sequence field byte-for-byte,
|
||||
and enforced uniqueness. Thirty-two concurrent mixed event writers produced a
|
||||
gap-free allocator stream. Invalid native finalization exhausted its
|
||||
three-attempt budget into named recovery without touching issue status. The
|
||||
legacy snapshot remained byte-equivalent with zero native history rows.
|
||||
three-attempt budget into named recovery without touching issue status. An
|
||||
actual flag-off heartbeat executed the legacy adapter/finalizer, remained
|
||||
byte-equivalent across native reconciliation, and created zero native rows.
|
||||
|
||||
# Credential, governance, budget, and legacy boundaries
|
||||
|
||||
@@ -133,7 +149,8 @@ legacy snapshot remained byte-equivalent with zero native history rows.
|
||||
proves the model envelope omits company/run/agent bindings and credential
|
||||
references. The native branch never creates a local agent JWT, MCP gateway, raw
|
||||
adapter environment, or legacy context. Runtime requests cannot auto-approve;
|
||||
the native interaction bridge rejects unsupported requests explicitly.
|
||||
the native interaction bridge materializes supported typed responses through
|
||||
the authorized service and rejects unsupported/governed/self-approved paths.
|
||||
|
||||
Explicit cancellation, agent pause, and budget hard-stop share
|
||||
`cancelRunInternal`, which now invokes the run-scoped normalized native session
|
||||
|
||||
@@ -40,21 +40,30 @@ existing server authority and keep legacy execution as the default.
|
||||
10. Treat model citations as claims. Only server-verifiable durable records can
|
||||
satisfy completion criteria, and every non-terminal decision must create or
|
||||
bind a real review, wake, blocker, or recovery path in the status transaction.
|
||||
11. Persist the closed native input before opening the provider. Result-less
|
||||
transport failures keep the same run live behind a bounded database lease;
|
||||
restart, flag disablement, and an already-active provider turn cannot create
|
||||
a second provider session, run, or turn.
|
||||
12. Project supported resolved interaction responses through the existing
|
||||
authorized issue-interaction service. Governed, unresolved, unsupported,
|
||||
cross-boundary, or self-approved requests fail closed without credentials.
|
||||
|
||||
# Evidence
|
||||
|
||||
- [Phase 6 verification](../evidence/2026-08-09-phase-06-verification.md)
|
||||
- [Phase 6 reference](../../docs/phase-06-thin-paperclip-adapter.md)
|
||||
- [Runnable tutorial](../../docs/tutorials/phase-06-thin-paperclip-adapter.md)
|
||||
- Package conformance and recovery: four files, six tests passed.
|
||||
- Phase 6 acceptance-matrix entry points: ten files, twenty-eight tests passed
|
||||
- Package conformance and recovery: four files, seven tests passed.
|
||||
- Phase 6 acceptance-matrix entry points: thirteen files, thirty-nine tests passed
|
||||
with zero skips, including the database-backed selected-task canary, six
|
||||
atomic-liveness failpoints, migration, sequencing, bounded recovery, and
|
||||
legacy compatibility.
|
||||
- Focused database rehearsal: five files, eleven tests passed with zero skips;
|
||||
the legacy event/read snapshot was byte-equivalent and had zero native rows.
|
||||
- Section 18.13 source corpus: 52 fixtures, all 70 matrix rows, and 18 structural
|
||||
checks passed with every fixture bound to a named consumer.
|
||||
- The migration rehearsal rewound to a production-shaped pre-0211 schema and
|
||||
applied the complete migration; the actual flag-off heartbeat executed the
|
||||
legacy adapter/finalizer and created zero native rows.
|
||||
- Section 18.13 source corpus: 52 fixture executions and 70 unique matrix-row
|
||||
executions passed through production consumers with observed result/effect
|
||||
digests; expected-object hashing and import-only aliases were removed.
|
||||
- Runner, shared, server, and database typechecks passed; migration safety
|
||||
passed.
|
||||
|
||||
@@ -75,14 +84,16 @@ existing server authority and keep legacy execution as the default.
|
||||
|
||||
- Phase 6 supports local, standard, workspace-backed `codex_local` tasks only.
|
||||
- It adds no browser UI or public runner WebSocket.
|
||||
- Typed provider requests are rejected unless an existing Paperclip governance
|
||||
path can materialize an authorized response; there is no auto-approval.
|
||||
- Resolved confirmations and question answers are materialized only through the
|
||||
existing authorized Paperclip interaction service; governed and unsupported
|
||||
requests cannot auto-approve.
|
||||
- Live operator proof depends on a board-authorized local instance and an
|
||||
already authenticated Codex installation; deterministic CI uses the same
|
||||
public session contract with a scripted backend.
|
||||
- The remediation did not dispatch a new live provider task or mutate rollout
|
||||
settings. The verified “internal canary” and post-kill-switch trace are the
|
||||
real embedded-PostgreSQL selected-task and legacy-snapshot cases; the live
|
||||
embedded-PostgreSQL scripted selected-task and actual flag-off legacy
|
||||
heartbeat cases; the live
|
||||
tutorial remains an explicit operator checkpoint.
|
||||
|
||||
# Follow-up questions
|
||||
|
||||
@@ -19,6 +19,11 @@ describe("ControlPlanePort conformance", () => {
|
||||
highestContiguousSourceSeq: 3,
|
||||
duplicateDisposition: "duplicate",
|
||||
terminalReplayIdempotent: true,
|
||||
openBindingRejected: true,
|
||||
eventIdMutationRejected: true,
|
||||
eventSequenceMutationRejected: true,
|
||||
replayBindingRejected: true,
|
||||
resultMutationRejected: true,
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -10,6 +10,11 @@ export interface ControlPlanePortConformanceSnapshot {
|
||||
highestContiguousSourceSeq: number;
|
||||
duplicateDisposition: "duplicate";
|
||||
terminalReplayIdempotent: boolean;
|
||||
openBindingRejected: boolean;
|
||||
eventIdMutationRejected: boolean;
|
||||
eventSequenceMutationRejected: boolean;
|
||||
replayBindingRejected: boolean;
|
||||
resultMutationRejected: boolean;
|
||||
}
|
||||
|
||||
export interface ControlPlanePortConformanceHarness {
|
||||
@@ -83,6 +88,16 @@ export async function runControlPlanePortConformance(
|
||||
await harness.start?.();
|
||||
try {
|
||||
await harness.port.openRun(CONTROL_PLANE_CONFORMANCE_OPEN);
|
||||
let openBindingRejected = false;
|
||||
try {
|
||||
await harness.port.openRun({
|
||||
...CONTROL_PLANE_CONFORMANCE_OPEN,
|
||||
identity: { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, issueId: "forged-issue" },
|
||||
});
|
||||
} catch {
|
||||
openBindingRejected = true;
|
||||
}
|
||||
if (!openBindingRejected) throw new Error("mismatched run binding was accepted");
|
||||
const first = await harness.port.appendEvent(CONTROL_PLANE_CONFORMANCE_EVENTS[0]);
|
||||
if (first.disposition !== "committed" || first.highestContiguousSourceSeq !== 1) {
|
||||
throw new Error("first event did not commit at cursor 1");
|
||||
@@ -93,6 +108,27 @@ export async function runControlPlanePortConformance(
|
||||
if (recovered.highestContiguousSourceSeq !== 3) throw new Error("source gap did not recover");
|
||||
const duplicate = await harness.port.appendEvent(CONTROL_PLANE_CONFORMANCE_EVENTS[1]);
|
||||
if (duplicate.disposition !== "duplicate") throw new Error("identical duplicate was not idempotent");
|
||||
let eventIdMutationRejected = false;
|
||||
try {
|
||||
await harness.port.appendEvent({
|
||||
...CONTROL_PLANE_CONFORMANCE_EVENTS[1],
|
||||
payload: { mutated: true },
|
||||
});
|
||||
} catch {
|
||||
eventIdMutationRejected = true;
|
||||
}
|
||||
if (!eventIdMutationRejected) throw new Error("mutated event id replay was accepted");
|
||||
let eventSequenceMutationRejected = false;
|
||||
try {
|
||||
await harness.port.appendEvent({
|
||||
...CONTROL_PLANE_CONFORMANCE_EVENTS[1],
|
||||
sourceEventId: "runner-phase6-conformance:mutated-sequence-two",
|
||||
payload: { mutated: true },
|
||||
});
|
||||
} catch {
|
||||
eventSequenceMutationRejected = true;
|
||||
}
|
||||
if (!eventSequenceMutationRejected) throw new Error("mutated source sequence replay was accepted");
|
||||
|
||||
const replay = await harness.port.replayEvents({
|
||||
runId: CONTROL_PLANE_CONFORMANCE_OPEN.identity.runId,
|
||||
@@ -103,6 +139,18 @@ export async function runControlPlanePortConformance(
|
||||
if (replay.events.map((entry) => entry.sourceSeq).join(",") !== "2,3") {
|
||||
throw new Error("exclusive replay cursor returned the wrong event set");
|
||||
}
|
||||
let replayBindingRejected = false;
|
||||
try {
|
||||
await harness.port.replayEvents({
|
||||
runId: "forged-run",
|
||||
sourceInstanceId: "runner-phase6-conformance",
|
||||
afterSourceSeq: 0,
|
||||
limit: 10,
|
||||
});
|
||||
} catch {
|
||||
replayBindingRejected = true;
|
||||
}
|
||||
if (!replayBindingRejected) throw new Error("mismatched replay binding was accepted");
|
||||
|
||||
const completion: CompleteControlPlaneRunInput = {
|
||||
result: CONTROL_PLANE_CONFORMANCE_RESULT,
|
||||
@@ -113,11 +161,26 @@ export async function runControlPlanePortConformance(
|
||||
};
|
||||
await harness.port.completeRun(completion);
|
||||
await harness.port.completeRun(completion);
|
||||
let resultMutationRejected = false;
|
||||
try {
|
||||
await harness.port.completeRun({
|
||||
...completion,
|
||||
result: { ...completion.result, summary: "mutated result replay" },
|
||||
});
|
||||
} catch {
|
||||
resultMutationRejected = true;
|
||||
}
|
||||
if (!resultMutationRejected) throw new Error("mutated result replay was accepted");
|
||||
return {
|
||||
eventCount: replay.highestContiguousSourceSeq,
|
||||
highestContiguousSourceSeq: replay.highestContiguousSourceSeq,
|
||||
duplicateDisposition: duplicate.disposition,
|
||||
terminalReplayIdempotent: true,
|
||||
openBindingRejected,
|
||||
eventIdMutationRejected,
|
||||
eventSequenceMutationRejected,
|
||||
replayBindingRejected,
|
||||
resultMutationRejected,
|
||||
};
|
||||
} finally {
|
||||
await harness.stop?.();
|
||||
|
||||
@@ -101,6 +101,98 @@ function highestContiguous(events: PrpEvent[]): number {
|
||||
}
|
||||
|
||||
describe("executeNativeSession recovery", () => {
|
||||
it("continues a provider-reported active turn without starting a duplicate turn", async () => {
|
||||
const checkpoint: PersistedNativeSession = {
|
||||
backendKind: "mock",
|
||||
sessionId: "driver-recovery",
|
||||
identity,
|
||||
providerSessionId: "provider-recovery",
|
||||
cursor: "0",
|
||||
activeTurnId: null,
|
||||
pendingRuntimeRequests: [],
|
||||
lineage: [],
|
||||
};
|
||||
const providerSnapshot: PersistedNativeSession = {
|
||||
...checkpoint,
|
||||
cursor: "1",
|
||||
activeTurnId: "turn-recovery",
|
||||
};
|
||||
const terminalEvent: PrpEvent = {
|
||||
schema: "paperclip.prp.event.v1",
|
||||
sourceEventId: "provider-recovery:1",
|
||||
sourceSeq: 1,
|
||||
sourceInstanceId: "provider-recovery",
|
||||
sourceKind: "provider",
|
||||
runId: identity.runId,
|
||||
normalizedSessionId: identity.sessionId,
|
||||
turnId: "turn-recovery",
|
||||
eventType: "turn.completed",
|
||||
schemaVersion: 1,
|
||||
priority: 0,
|
||||
emittedAt: "2026-08-09T00:00:00.000Z",
|
||||
payload: {},
|
||||
};
|
||||
const bySource = new Map<string, PrpEvent[]>();
|
||||
const startTurn = vi.fn(async () => ({ turnId: "duplicate-turn" }));
|
||||
const openSession = vi.fn(async () => { throw new Error("must recover the provider session"); });
|
||||
const session: NativeSession = {
|
||||
identity: () => identity,
|
||||
async capabilities() {
|
||||
return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true };
|
||||
},
|
||||
async *events() { yield terminalEvent; },
|
||||
startTurn,
|
||||
async result() { return { result, terminal, turnId: "turn-recovery" }; },
|
||||
async snapshot() { return structuredClone(providerSnapshot); },
|
||||
async close() {},
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
async descriptor() {
|
||||
return {
|
||||
kind: "mock",
|
||||
name: "recovery-backend",
|
||||
version: "1",
|
||||
capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true },
|
||||
};
|
||||
},
|
||||
openSession,
|
||||
async recoverSession() { return { recovered: true, session }; },
|
||||
};
|
||||
const port: ControlPlanePort = {
|
||||
async openRun() {},
|
||||
async loadSessionCheckpoint() { return structuredClone(checkpoint); },
|
||||
async checkpointSession() {},
|
||||
async appendEvent(event) {
|
||||
const list = bySource.get(event.sourceInstanceId) ?? [];
|
||||
list.push(structuredClone(event));
|
||||
bySource.set(event.sourceInstanceId, list);
|
||||
return {
|
||||
cursor: list.length,
|
||||
highestContiguousSourceSeq: highestContiguous(list),
|
||||
disposition: "committed",
|
||||
};
|
||||
},
|
||||
async replayEvents(replay) {
|
||||
const list = bySource.get(replay.sourceInstanceId) ?? [];
|
||||
return {
|
||||
events: structuredClone(list.filter((event) => event.sourceSeq > replay.afterSourceSeq)),
|
||||
highestContiguousSourceSeq: highestContiguous(list),
|
||||
};
|
||||
},
|
||||
async completeRun() {},
|
||||
};
|
||||
|
||||
await expect(executeNativeSession({
|
||||
input,
|
||||
backend,
|
||||
controlPlane: port,
|
||||
runnerInstanceId: "runner-recovery",
|
||||
controlPlaneInstanceId: "control-recovery",
|
||||
})).resolves.toMatchObject({ turnId: "turn-recovery", providerSessionId: "provider-recovery" });
|
||||
expect(openSession).not.toHaveBeenCalled();
|
||||
expect(startTurn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("recovers a completed checkpoint and appends only a missing control terminal fact", async () => {
|
||||
const checkpoint: PersistedNativeSession = {
|
||||
backendKind: "mock",
|
||||
|
||||
@@ -120,19 +120,24 @@ export async function executeNativeSession(options: ExecuteNativeSessionOptions)
|
||||
await options.controlPlane.checkpointSession(await session.snapshot());
|
||||
}
|
||||
};
|
||||
await checkpoint();
|
||||
const recoveredSnapshot = await session.snapshot();
|
||||
await options.controlPlane.checkpointSession?.(recoveredSnapshot);
|
||||
|
||||
let consumed = { event: null as PrpEvent | null, eventCount: 0, highestContiguousSourceSeq: 0 };
|
||||
let completed = persistedSession?.semanticResult && persistedSession.terminal
|
||||
const completionSnapshot = recoveredSnapshot.semanticResult && recoveredSnapshot.terminal
|
||||
? recoveredSnapshot
|
||||
: persistedSession;
|
||||
let completed = completionSnapshot?.semanticResult && completionSnapshot.terminal
|
||||
? {
|
||||
result: persistedSession.semanticResult,
|
||||
terminal: persistedSession.terminal,
|
||||
turnId: persistedSession.activeTurnId ?? null,
|
||||
result: completionSnapshot.semanticResult,
|
||||
terminal: completionSnapshot.terminal,
|
||||
turnId: completionSnapshot.activeTurnId ?? null,
|
||||
}
|
||||
: null;
|
||||
if (!completed) {
|
||||
const consuming = consumeTurn(session, options.controlPlane, options.timeoutMs ?? 900_000);
|
||||
if (!recovered || !persistedSession?.activeTurnId) {
|
||||
const recoveredActiveTurnId = recoveredSnapshot.activeTurnId ?? persistedSession?.activeTurnId ?? null;
|
||||
if (!recovered || !recoveredActiveTurnId) {
|
||||
await session.startTurn({
|
||||
message: { role: "user", text: JSON.stringify(buildNativeModelEnvelope(input)) },
|
||||
});
|
||||
@@ -145,8 +150,8 @@ export async function executeNativeSession(options: ExecuteNativeSessionOptions)
|
||||
}
|
||||
if (completed === null) throw new Error("native_finalization_missing: session returned no semantic result");
|
||||
let terminal: PrpTerminalState;
|
||||
if (persistedSession?.semanticResult && persistedSession.terminal) {
|
||||
terminal = persistedSession.terminal;
|
||||
if (completionSnapshot?.semanticResult && completionSnapshot.terminal) {
|
||||
terminal = completionSnapshot.terminal;
|
||||
} else {
|
||||
terminal = terminalFromEvent(consumed.event!, completed.result.reportedWorkDisposition);
|
||||
}
|
||||
|
||||
@@ -1,80 +1,139 @@
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
completionContracts,
|
||||
createDb,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
nativeRunFinalizations,
|
||||
nativeRunResults,
|
||||
projectWorkspaces,
|
||||
projects,
|
||||
statusDecisions,
|
||||
workAssessments,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js";
|
||||
|
||||
const adapterExecute = vi.hoisted(() => vi.fn(async () => ({
|
||||
exitCode: 0,
|
||||
signal: null,
|
||||
timedOut: false,
|
||||
summary: "Legacy adapter completed through the flag-off heartbeat.",
|
||||
resultJson: { summary: "Legacy bytes", nested: { count: 1, ok: true } },
|
||||
provider: "test",
|
||||
model: "legacy-test",
|
||||
})));
|
||||
|
||||
vi.mock("../adapters/index.js", () => ({
|
||||
getServerAdapter: () => ({
|
||||
type: "codex_local",
|
||||
execute: adapterExecute,
|
||||
supportsLocalAgentJwt: false,
|
||||
}),
|
||||
findActiveServerAdapter: () => ({
|
||||
type: "codex_local",
|
||||
execute: adapterExecute,
|
||||
supportsLocalAgentJwt: false,
|
||||
}),
|
||||
listAdapterModelProfiles: async () => [],
|
||||
runningProcesses: new Map(),
|
||||
}));
|
||||
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js";
|
||||
|
||||
describe("P6-32 legacy finalization regression", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
const companyId = "71000000-0000-4000-8000-000000000001";
|
||||
const agentId = "71000000-0000-4000-8000-000000000002";
|
||||
const runId = "71000000-0000-4000-8000-000000000003";
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const projectId = randomUUID();
|
||||
const projectWorkspaceId = randomUUID();
|
||||
const issueId = randomUUID();
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-legacy-");
|
||||
db = createDb(temporary.connectionString);
|
||||
await db.insert(companies).values({ id: companyId, name: "Legacy snapshot", issuePrefix: "LGC" });
|
||||
await instanceSettingsService(db).updateExperimental({ enableNativeRunner: false });
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Legacy snapshot",
|
||||
issuePrefix: "LGC",
|
||||
status: "active",
|
||||
defaultResponsibleUserId: "responsible-user",
|
||||
});
|
||||
await db.insert(projects).values({ id: projectId, companyId, name: "Legacy project", status: "active" });
|
||||
await db.insert(projectWorkspaces).values({
|
||||
id: projectWorkspaceId,
|
||||
companyId,
|
||||
projectId,
|
||||
name: "Primary",
|
||||
cwd: fileURLToPath(new URL("../../../", import.meta.url)),
|
||||
isPrimary: true,
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Legacy agent",
|
||||
adapterType: "codex_local",
|
||||
status: "idle",
|
||||
runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } },
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "succeeded",
|
||||
runtimeMode: "legacy",
|
||||
nextEventSeq: 8,
|
||||
exitCode: 0,
|
||||
resultJson: { summary: "Legacy bytes", nested: { count: 1, ok: true } },
|
||||
contextSnapshot: { issueId: "legacy-opaque-issue" },
|
||||
});
|
||||
await db.insert(heartbeatRunEvents).values({
|
||||
companyId,
|
||||
runId,
|
||||
agentId,
|
||||
seq: 7,
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
message: "Legacy event bytes",
|
||||
payload: { state: "completed", nested: [1, "two", false] },
|
||||
projectId,
|
||||
projectWorkspaceId,
|
||||
title: "Complete through the legacy adapter",
|
||||
status: "in_progress",
|
||||
workMode: "standard",
|
||||
assigneeAgentId: agentId,
|
||||
});
|
||||
}, 30_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await temporary.cleanup();
|
||||
if (temporary) {
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeatService(db));
|
||||
await temporary.cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps the post-kill-switch legacy read snapshot byte-equivalent with zero native rows", async () => {
|
||||
const beforeRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const beforeEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
||||
const beforeBytes = JSON.stringify({ run: beforeRun, events: beforeEvents });
|
||||
it("executes a flag-off heartbeat through the legacy adapter with byte-stable reads and zero native rows", async () => {
|
||||
const heartbeat = heartbeatService(db);
|
||||
const queued = await heartbeat.wakeup(agentId, {
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_commented",
|
||||
payload: { issueId },
|
||||
contextSnapshot: { issueId, taskId: issueId, skipIssueComment: true },
|
||||
});
|
||||
expect(queued).not.toBeNull();
|
||||
await drainHeartbeatRunsToQuiescence(db, heartbeat);
|
||||
const runId = queued!.id;
|
||||
const publicBefore = await heartbeat.getRun(runId);
|
||||
|
||||
expect(adapterExecute).toHaveBeenCalledOnce();
|
||||
expect(publicBefore).toMatchObject({
|
||||
id: runId,
|
||||
status: "succeeded",
|
||||
runtimeMode: "legacy",
|
||||
runtimeModeReason: "instance_flag_disabled",
|
||||
resultJson: { summary: "Legacy bytes", nested: { count: 1, ok: true } },
|
||||
});
|
||||
await expect(reconcileNativeFinalizations(db, [runId])).resolves.toEqual([]);
|
||||
expect(JSON.stringify(await heartbeat.getRun(runId))).toBe(JSON.stringify(publicBefore));
|
||||
|
||||
const afterRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const afterEvents = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
||||
expect(JSON.stringify({ run: afterRun, events: afterEvents })).toBe(beforeBytes);
|
||||
await expect(db.select().from(completionContracts).where(eq(completionContracts.companyId, companyId))).resolves.toHaveLength(0);
|
||||
await expect(db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId))).resolves.toHaveLength(0);
|
||||
await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).resolves.toHaveLength(0);
|
||||
await expect(db.select().from(workAssessments).where(eq(workAssessments.runId, runId))).resolves.toHaveLength(0);
|
||||
await expect(db.select().from(statusDecisions)
|
||||
.where(eq(statusDecisions.companyId, companyId))).resolves.toHaveLength(0);
|
||||
});
|
||||
await expect(db.select().from(statusDecisions).where(eq(statusDecisions.companyId, companyId))).resolves.toHaveLength(0);
|
||||
await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))).resolves.toHaveLength(1);
|
||||
}, 30_000);
|
||||
});
|
||||
@@ -1,9 +1,9 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
applyPendingMigrations,
|
||||
createDb,
|
||||
heartbeatRunEvents,
|
||||
heartbeatRuns,
|
||||
@@ -13,38 +13,81 @@ import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.j
|
||||
describe("P6-18 / MIG-01..04 native finalization migration", () => {
|
||||
it("repairs only later duplicates and preserves legacy event bytes and cursors", async () => {
|
||||
const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-migration-");
|
||||
const db = createDb(temporary.connectionString);
|
||||
const migration = await readFile(
|
||||
new URL("../../../packages/db/src/migrations/0211_famous_guardsmen.sql", import.meta.url),
|
||||
"utf8",
|
||||
);
|
||||
const migrationHash = createHash("sha256").update(migration).digest("hex");
|
||||
const rawDb = createDb(temporary.connectionString);
|
||||
try {
|
||||
const companyId = "10000000-0000-4000-8000-000000000001";
|
||||
const agentId = "10000000-0000-4000-8000-000000000002";
|
||||
const runId = "10000000-0000-4000-8000-000000000003";
|
||||
await db.insert(companies).values({ id: companyId, name: "Migration fixture", issuePrefix: "MIG" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Migration agent" });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", nextEventSeq: 1 });
|
||||
await db.execute(sql.raw("DROP INDEX heartbeat_run_events_run_seq_uq"));
|
||||
await db.insert(heartbeatRunEvents).values([
|
||||
{ companyId, runId, agentId, seq: 1, eventType: "legacy.start", stream: "system", level: "info", message: "one", payload: { bytes: "α-1" }, createdAt: new Date("2026-08-01T00:00:01.000Z") },
|
||||
{ companyId, runId, agentId, seq: 5, eventType: "legacy.log", stream: "stdout", level: "info", message: "first-five", payload: { bytes: "β-5a" }, createdAt: new Date("2026-08-01T00:00:02.000Z") },
|
||||
{ companyId, runId, agentId, seq: 5, eventType: "legacy.log", stream: "stderr", level: "warn", message: "duplicate-five", payload: { bytes: "γ-5b" }, createdAt: new Date("2026-08-01T00:00:03.000Z") },
|
||||
{ companyId, runId, agentId, seq: 9, eventType: "legacy.end", stream: "system", level: "info", message: "nine", payload: { bytes: "δ-9" }, createdAt: new Date("2026-08-01T00:00:04.000Z") },
|
||||
]);
|
||||
// Reconstruct the actual pre-0211 shape rather than extracting selected
|
||||
// repair statements from the migration under test.
|
||||
await rawDb.execute(sql.raw(`
|
||||
DROP TABLE IF EXISTS status_decision_effects, status_decisions, work_assessments,
|
||||
native_run_finalizations, native_run_results, completion_contracts CASCADE;
|
||||
DROP TRIGGER IF EXISTS paperclip_issue_status_version_trigger ON issues;
|
||||
DROP FUNCTION IF EXISTS paperclip_bump_issue_status_version();
|
||||
DROP INDEX IF EXISTS heartbeat_run_events_run_seq_uq;
|
||||
DROP INDEX IF EXISTS heartbeat_run_events_run_source_event_uq;
|
||||
DROP INDEX IF EXISTS heartbeat_run_events_run_source_seq_uq;
|
||||
ALTER TABLE heartbeat_run_events
|
||||
DROP COLUMN IF EXISTS source_instance_id,
|
||||
DROP COLUMN IF EXISTS source_event_id,
|
||||
DROP COLUMN IF EXISTS source_seq,
|
||||
DROP COLUMN IF EXISTS source_payload_sha256,
|
||||
DROP COLUMN IF EXISTS protocol_schema_version;
|
||||
ALTER TABLE heartbeat_run_events ALTER COLUMN seq TYPE integer;
|
||||
ALTER TABLE heartbeat_runs
|
||||
DROP COLUMN IF EXISTS runtime_mode,
|
||||
DROP COLUMN IF EXISTS runtime_mode_resolver_version,
|
||||
DROP COLUMN IF EXISTS runtime_mode_reason,
|
||||
DROP COLUMN IF EXISTS runtime_mode_resolved_at,
|
||||
DROP COLUMN IF EXISTS runner_profile_json,
|
||||
DROP COLUMN IF EXISTS runner_instance_id,
|
||||
DROP COLUMN IF EXISTS native_session_id,
|
||||
DROP COLUMN IF EXISTS driver_kind,
|
||||
DROP COLUMN IF EXISTS driver_version,
|
||||
DROP COLUMN IF EXISTS completion_contract_id,
|
||||
DROP COLUMN IF EXISTS completion_contract_sha256,
|
||||
DROP COLUMN IF EXISTS next_event_seq,
|
||||
DROP COLUMN IF EXISTS native_phase,
|
||||
DROP COLUMN IF EXISTS native_phase_updated_at;
|
||||
ALTER TABLE issues
|
||||
DROP COLUMN IF EXISTS status_version,
|
||||
DROP COLUMN IF EXISTS last_status_decision_id;
|
||||
`));
|
||||
await rawDb.execute(sql`DELETE FROM "drizzle"."__drizzle_migrations" WHERE "hash" = ${migrationHash}`);
|
||||
await rawDb.execute(sql`
|
||||
INSERT INTO companies (id, name, issue_prefix)
|
||||
VALUES (${companyId}, 'Migration fixture', 'MIG')
|
||||
`);
|
||||
await rawDb.execute(sql`
|
||||
INSERT INTO agents (id, company_id, name)
|
||||
VALUES (${agentId}, ${companyId}, 'Migration agent')
|
||||
`);
|
||||
await rawDb.execute(sql`
|
||||
INSERT INTO heartbeat_runs (id, company_id, agent_id, status)
|
||||
VALUES (${runId}, ${companyId}, ${agentId}, 'succeeded')
|
||||
`);
|
||||
await rawDb.execute(sql`
|
||||
INSERT INTO heartbeat_run_events
|
||||
(company_id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at)
|
||||
VALUES
|
||||
(${companyId}, ${runId}, ${agentId}, 1, 'legacy.start', 'system', 'info', 'one', ${JSON.stringify({ bytes: "α-1" })}::jsonb, '2026-08-01T00:00:01.000Z'),
|
||||
(${companyId}, ${runId}, ${agentId}, 5, 'legacy.log', 'stdout', 'info', 'first-five', ${JSON.stringify({ bytes: "β-5a" })}::jsonb, '2026-08-01T00:00:02.000Z'),
|
||||
(${companyId}, ${runId}, ${agentId}, 5, 'legacy.log', 'stderr', 'warn', 'duplicate-five', ${JSON.stringify({ bytes: "γ-5b" })}::jsonb, '2026-08-01T00:00:03.000Z'),
|
||||
(${companyId}, ${runId}, ${agentId}, 9, 'legacy.end', 'system', 'info', 'nine', ${JSON.stringify({ bytes: "δ-9" })}::jsonb, '2026-08-01T00:00:04.000Z')
|
||||
`);
|
||||
const beforeResult = await rawDb.execute(sql`
|
||||
SELECT * FROM heartbeat_run_events WHERE run_id = ${runId} ORDER BY id
|
||||
`);
|
||||
const before = [...beforeResult] as unknown as Record<string, unknown>[];
|
||||
|
||||
const before = await db.select().from(heartbeatRunEvents)
|
||||
.where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.id);
|
||||
const migration = await readFile(
|
||||
new URL("../../../packages/db/src/migrations/0211_famous_guardsmen.sql", import.meta.url),
|
||||
"utf8",
|
||||
);
|
||||
const statements = migration.split("--> statement-breakpoint").map((entry) => entry.trim());
|
||||
const duplicateRepair = statements.find((entry) => entry.includes("WITH ranked AS"));
|
||||
const allocatorSeed = statements.find((entry) => entry.startsWith('UPDATE "heartbeat_runs" AS run'));
|
||||
expect(duplicateRepair).toBeTruthy();
|
||||
expect(allocatorSeed).toBeTruthy();
|
||||
await db.execute(sql.raw(duplicateRepair!));
|
||||
await db.execute(sql.raw(allocatorSeed!));
|
||||
await db.execute(sql.raw(
|
||||
'CREATE UNIQUE INDEX "heartbeat_run_events_run_seq_uq" ON "heartbeat_run_events" ("run_id", "seq")',
|
||||
));
|
||||
await applyPendingMigrations(temporary.connectionString);
|
||||
const db = createDb(temporary.connectionString);
|
||||
|
||||
const after = await db.select().from(heartbeatRunEvents)
|
||||
.where(eq(heartbeatRunEvents.runId, runId)).orderBy(heartbeatRunEvents.id);
|
||||
@@ -53,10 +96,22 @@ describe("P6-18 / MIG-01..04 native finalization migration", () => {
|
||||
.where(eq(heartbeatRuns.id, runId)))[0]?.nextEventSeq).toBe(11);
|
||||
|
||||
// The repaired duplicate's cursor is the only changed byte-equivalent read field.
|
||||
expect(after.map(({ seq: _seq, ...row }) => row)).toEqual(before.map(({ seq: _seq, ...row }) => row));
|
||||
expect(after[0]?.seq).toBe(before[0]?.seq);
|
||||
expect(after[1]?.seq).toBe(before[1]?.seq);
|
||||
expect(after[3]?.seq).toBe(before[3]?.seq);
|
||||
const legacyColumns = (row: Record<string, unknown>) => ({
|
||||
id: String(row.id),
|
||||
companyId: row.companyId ?? row.company_id,
|
||||
runId: row.runId ?? row.run_id,
|
||||
agentId: row.agentId ?? row.agent_id,
|
||||
eventType: row.eventType ?? row.event_type,
|
||||
stream: row.stream,
|
||||
level: row.level,
|
||||
message: row.message,
|
||||
payload: row.payload,
|
||||
createdAt: new Date(String(row.createdAt ?? row.created_at)).toISOString(),
|
||||
});
|
||||
expect(after.map((row) => legacyColumns(row))).toEqual(before.map(legacyColumns));
|
||||
expect(after[0]?.seq).toBe(Number(before[0]?.seq));
|
||||
expect(after[1]?.seq).toBe(Number(before[1]?.seq));
|
||||
expect(after[3]?.seq).toBe(Number(before[3]?.seq));
|
||||
await expect(db.insert(heartbeatRunEvents).values({
|
||||
companyId, runId, agentId, seq: 5, eventType: "must-conflict",
|
||||
})).rejects.toThrow();
|
||||
|
||||
@@ -0,0 +1,172 @@
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import {
|
||||
materializeNativeInteractionResponses,
|
||||
NativeInteractionBridgeError,
|
||||
} from "../services/native-runtime/native-interaction-bridge.js";
|
||||
|
||||
describe("P6-19 native interaction bridge", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
const companyId = "78000000-0000-4000-8000-000000000001";
|
||||
const agentId = "78000000-0000-4000-8000-000000000002";
|
||||
const issueId = "78000000-0000-4000-8000-000000000003";
|
||||
const runId = "78000000-0000-4000-8000-000000000004";
|
||||
const confirmationId = "78000000-0000-4000-8000-000000000005";
|
||||
const questionsId = "78000000-0000-4000-8000-000000000006";
|
||||
const governedId = "78000000-0000-4000-8000-000000000007";
|
||||
const selfApprovedId = "78000000-0000-4000-8000-000000000008";
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-interaction-");
|
||||
db = createDb(temporary.connectionString);
|
||||
await db.insert(companies).values({ id: companyId, name: "Native interaction", issuePrefix: "NIB" });
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Native interaction agent",
|
||||
adapterType: "codex_local",
|
||||
status: "running",
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Consume an authorized interaction response",
|
||||
status: "in_progress",
|
||||
assigneeAgentId: agentId,
|
||||
workMode: "standard",
|
||||
});
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "running",
|
||||
contextSnapshot: { issueId, interactionId: confirmationId },
|
||||
});
|
||||
await db.insert(issueThreadInteractions).values([
|
||||
{
|
||||
id: confirmationId,
|
||||
companyId,
|
||||
issueId,
|
||||
kind: "request_confirmation",
|
||||
status: "accepted",
|
||||
resolvedByUserId: "board-user",
|
||||
resolvedAt: new Date(),
|
||||
payload: { version: 1, prompt: "Continue?" },
|
||||
result: { version: 1, outcome: "accepted" },
|
||||
},
|
||||
{
|
||||
id: questionsId,
|
||||
companyId,
|
||||
issueId,
|
||||
kind: "ask_user_questions",
|
||||
status: "answered",
|
||||
resolvedByUserId: "board-user",
|
||||
resolvedAt: new Date(),
|
||||
payload: {
|
||||
version: 1,
|
||||
questions: [{
|
||||
id: "choice",
|
||||
prompt: "Which path?",
|
||||
selectionMode: "single",
|
||||
options: [{ id: "safe", label: "Safe" }],
|
||||
}],
|
||||
},
|
||||
result: { version: 1, answers: [{ questionId: "choice", optionIds: ["safe"] }] },
|
||||
},
|
||||
{
|
||||
id: governedId,
|
||||
companyId,
|
||||
issueId,
|
||||
kind: "request_confirmation",
|
||||
status: "accepted",
|
||||
resolvedByUserId: "board-user",
|
||||
resolvedAt: new Date(),
|
||||
payload: {
|
||||
version: 1,
|
||||
prompt: "Execute write?",
|
||||
toolAction: {
|
||||
version: 1,
|
||||
actionRequestId: "78000000-0000-4000-8000-000000000010",
|
||||
invocationId: "78000000-0000-4000-8000-000000000011",
|
||||
toolName: "write",
|
||||
toolDisplayName: "Write",
|
||||
connectionId: null,
|
||||
applicationId: null,
|
||||
appDisplayName: null,
|
||||
risk: "write",
|
||||
previewMarkdown: "write",
|
||||
argumentsSummaryJson: "{}",
|
||||
argumentsHash: "hash",
|
||||
expiresAt: new Date(Date.now() + 60_000).toISOString(),
|
||||
},
|
||||
},
|
||||
result: { version: 1, outcome: "accepted" },
|
||||
},
|
||||
{
|
||||
id: selfApprovedId,
|
||||
companyId,
|
||||
issueId,
|
||||
kind: "request_confirmation",
|
||||
status: "accepted",
|
||||
createdByAgentId: agentId,
|
||||
resolvedByAgentId: agentId,
|
||||
resolvedByRunId: runId,
|
||||
resolvedAt: new Date(),
|
||||
payload: { version: 1, prompt: "Self approve?" },
|
||||
result: { version: 1, outcome: "accepted" },
|
||||
},
|
||||
]);
|
||||
}, 30_000);
|
||||
|
||||
afterAll(async () => temporary?.cleanup());
|
||||
|
||||
it("projects supported typed responses through the authorized interaction service", async () => {
|
||||
await expect(materializeNativeInteractionResponses({
|
||||
db,
|
||||
companyId,
|
||||
issueId,
|
||||
runId,
|
||||
agentId,
|
||||
interactionIds: [questionsId, confirmationId],
|
||||
})).resolves.toEqual([
|
||||
{
|
||||
interactionId: confirmationId,
|
||||
kind: "request_confirmation",
|
||||
response: { status: "accepted", result: { version: 1, outcome: "accepted" } },
|
||||
},
|
||||
{
|
||||
interactionId: questionsId,
|
||||
kind: "ask_user_questions",
|
||||
response: {
|
||||
status: "answered",
|
||||
result: { version: 1, answers: [{ questionId: "choice", optionIds: ["safe"] }] },
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
[governedId, "native_interaction_governed_request_unsupported"],
|
||||
[selfApprovedId, "native_interaction_self_approval"],
|
||||
])("fails closed for governed or self-approved interaction %s", async (interactionId, code) => {
|
||||
const error = await materializeNativeInteractionResponses({
|
||||
db,
|
||||
companyId,
|
||||
issueId,
|
||||
runId,
|
||||
agentId,
|
||||
interactionIds: [interactionId],
|
||||
}).catch((caught) => caught);
|
||||
expect(error).toBeInstanceOf(NativeInteractionBridgeError);
|
||||
expect(error).toMatchObject({ code });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,184 @@
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
createDb,
|
||||
heartbeatRuns,
|
||||
issueRecoveryActions,
|
||||
issues,
|
||||
nativeRunFinalizations,
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import {
|
||||
claimNativeSessionResumptions,
|
||||
dispatchNativeSessionResumptions,
|
||||
} from "../services/native-runtime/native-finalization-reconciler.js";
|
||||
|
||||
describe("P6-25 pre-result native session recovery", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
const companyId = "79000000-0000-4000-8000-000000000001";
|
||||
const agentId = "79000000-0000-4000-8000-000000000002";
|
||||
const issueId = "79000000-0000-4000-8000-000000000003";
|
||||
const runId = "79000000-0000-4000-8000-000000000004";
|
||||
const cancelledRunId = "79000000-0000-4000-8000-000000000005";
|
||||
const exhaustedRunId = "79000000-0000-4000-8000-000000000006";
|
||||
const missingCheckpointRunId = "79000000-0000-4000-8000-000000000007";
|
||||
const initialRunId = "79000000-0000-4000-8000-000000000008";
|
||||
const persistedProfile = {
|
||||
mode: "native",
|
||||
nativeExecutionInput: { schema: "paperclip.native-execution-input.v1", binding: { runId } },
|
||||
sessionCheckpoint: {
|
||||
backendKind: "codex_app_server",
|
||||
sessionId: "persisted-session",
|
||||
identity: { runId },
|
||||
providerSessionId: "provider-session",
|
||||
activeTurnId: "active-turn",
|
||||
},
|
||||
};
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-resume-");
|
||||
db = createDb(temporary.connectionString);
|
||||
await db.insert(companies).values({ id: companyId, name: "Native resume", issuePrefix: "NRR" });
|
||||
await db.insert(agents).values({
|
||||
id: agentId,
|
||||
companyId,
|
||||
name: "Native resume agent",
|
||||
adapterType: "codex_local",
|
||||
status: "running",
|
||||
});
|
||||
await db.insert(issues).values({
|
||||
id: issueId,
|
||||
companyId,
|
||||
title: "Resume the same native run",
|
||||
status: "in_progress",
|
||||
assigneeAgentId: agentId,
|
||||
workMode: "standard",
|
||||
});
|
||||
await db.insert(heartbeatRuns).values([
|
||||
{
|
||||
id: runId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "running",
|
||||
runtimeMode: "native",
|
||||
runtimeModeResolvedAt: new Date(),
|
||||
runnerProfileJson: persistedProfile,
|
||||
contextSnapshot: { issueId },
|
||||
},
|
||||
{
|
||||
id: cancelledRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "cancelled",
|
||||
runtimeMode: "native",
|
||||
runtimeModeResolvedAt: new Date(),
|
||||
runnerProfileJson: persistedProfile,
|
||||
contextSnapshot: { issueId },
|
||||
},
|
||||
{
|
||||
id: exhaustedRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "failed",
|
||||
runtimeMode: "native",
|
||||
runtimeModeResolvedAt: new Date(),
|
||||
runnerProfileJson: persistedProfile,
|
||||
contextSnapshot: { issueId },
|
||||
},
|
||||
{
|
||||
id: missingCheckpointRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "running",
|
||||
runtimeMode: "native",
|
||||
runtimeModeResolvedAt: new Date(),
|
||||
runnerProfileJson: { nativeExecutionInput: persistedProfile.nativeExecutionInput },
|
||||
contextSnapshot: { issueId },
|
||||
},
|
||||
{
|
||||
id: initialRunId,
|
||||
companyId,
|
||||
agentId,
|
||||
status: "running",
|
||||
runtimeMode: "native",
|
||||
runtimeModeResolvedAt: new Date(),
|
||||
runnerProfileJson: { nativeExecutionInput: persistedProfile.nativeExecutionInput },
|
||||
contextSnapshot: { issueId },
|
||||
},
|
||||
]);
|
||||
await db.insert(nativeRunFinalizations).values([
|
||||
{ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) },
|
||||
{ runId: cancelledRunId, companyId, issueId, phase: "retryable_failure", attempt: 1, nextAttemptAt: new Date(0) },
|
||||
{ runId: exhaustedRunId, companyId, issueId, phase: "terminal_failure", attempt: 3 },
|
||||
{ runId: missingCheckpointRunId, companyId, issueId, phase: "observed", attempt: 1 },
|
||||
{ runId: initialRunId, companyId, issueId, phase: "observed", attempt: 0 },
|
||||
]);
|
||||
}, 30_000);
|
||||
|
||||
afterAll(async () => temporary?.cleanup());
|
||||
|
||||
it("wins one database lease for the original result-less run without consulting the flag", async () => {
|
||||
const results = await Promise.all([
|
||||
claimNativeSessionResumptions({ db, runnerInstanceId: "reaper-a", runIds: [runId] }),
|
||||
claimNativeSessionResumptions({ db, runnerInstanceId: "reaper-b", runIds: [runId] }),
|
||||
]);
|
||||
expect(results.flat()).toHaveLength(1);
|
||||
expect(results.flat()[0]).toMatchObject({ runId });
|
||||
await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))).resolves.toEqual([
|
||||
expect.objectContaining({ id: runId, status: "running", runtimeMode: "native" }),
|
||||
]);
|
||||
await expect(db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).resolves.toEqual([
|
||||
expect.objectContaining({ runId, phase: "observed", resultId: null, attempt: 1 }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("dispatches the persisted run id and lease to the live same-run resume consumer", async () => {
|
||||
await db.update(nativeRunFinalizations).set({
|
||||
phase: "retryable_failure",
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
nextAttemptAt: new Date(0),
|
||||
}).where(eq(nativeRunFinalizations.runId, runId));
|
||||
const dispatched: Array<{ runId: string; leaseOwner: string }> = [];
|
||||
await expect(dispatchNativeSessionResumptions({
|
||||
db,
|
||||
runnerInstanceId: "heartbeat-reaper",
|
||||
runIds: [runId],
|
||||
dispatch: (claim) => dispatched.push(claim),
|
||||
})).resolves.toHaveLength(1);
|
||||
expect(dispatched).toEqual([{ runId, leaseOwner: expect.stringContaining("heartbeat-reaper:resume:") }]);
|
||||
await expect(db.select().from(heartbeatRuns)).resolves.toHaveLength(5);
|
||||
await expect(db.select().from(nativeRunFinalizations)).resolves.toHaveLength(5);
|
||||
});
|
||||
|
||||
it("does not resume cancelled or exhausted runs and fails closed without a checkpoint", async () => {
|
||||
await expect(claimNativeSessionResumptions({
|
||||
db,
|
||||
runnerInstanceId: "reaper",
|
||||
runIds: [cancelledRunId, exhaustedRunId, missingCheckpointRunId],
|
||||
})).resolves.toEqual([]);
|
||||
await expect(db.select().from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, missingCheckpointRunId))).resolves.toEqual([
|
||||
expect.objectContaining({ phase: "terminal_failure", failureCode: "native_session_checkpoint_missing" }),
|
||||
]);
|
||||
await expect(db.select().from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, issueId))).resolves.toEqual([
|
||||
expect.objectContaining({ cause: "native_session_checkpoint_missing", wakePolicy: null }),
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not mistake the pre-first-attempt observed coordinator for an orphan", async () => {
|
||||
await expect(claimNativeSessionResumptions({
|
||||
db,
|
||||
runnerInstanceId: "reaper",
|
||||
runIds: [initialRunId],
|
||||
})).resolves.toEqual([]);
|
||||
await expect(db.select().from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, initialRunId))).resolves.toEqual([
|
||||
expect.objectContaining({ phase: "observed", attempt: 0, failureCode: null }),
|
||||
]);
|
||||
});
|
||||
});
|
||||
@@ -2,34 +2,34 @@ import { createHash } from "node:crypto";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { NativeEvidenceAssessment } from "../services/native-runtime/evidence-classifier.js";
|
||||
import { rejectUnsupportedNativeRuntimeRequest } from "../services/native-runtime/native-interaction-bridge.js";
|
||||
import { nativeSessionFailureDisposition } from "../services/native-runtime/native-session-executor.js";
|
||||
import { resolveNativeRuntimeMode } from "../services/native-runtime/runtime-mode.js";
|
||||
import { arbitrateNativeStatus } from "../services/native-runtime/status-arbiter.js";
|
||||
|
||||
type Fixture = {
|
||||
id: string;
|
||||
mode: "native" | "legacy";
|
||||
covers: Record<string, string[]>;
|
||||
expected: Record<string, unknown>;
|
||||
given: Record<string, unknown>;
|
||||
expected: {
|
||||
statusAction: string;
|
||||
reasonCode: string | null;
|
||||
requiredEffects: string[];
|
||||
forbiddenEffects: string[];
|
||||
livePathKind: string | null;
|
||||
nativeRecords: boolean;
|
||||
};
|
||||
};
|
||||
|
||||
type Corpus = {
|
||||
schema: string;
|
||||
corpusRevision: number;
|
||||
fixtures: Fixture[];
|
||||
};
|
||||
type Corpus = { schema: string; corpusRevision: number; fixtures: Fixture[] };
|
||||
|
||||
const corpusPath = fileURLToPath(new URL(
|
||||
"../../../packages/paperclip-runner/spec/fixtures/status-authority-phase5.json",
|
||||
import.meta.url,
|
||||
));
|
||||
|
||||
const consumerByMatrixPrefix: Record<string, string> = {
|
||||
SD: "server/native-status-arbiter-and-committer",
|
||||
TC: "package/native-session-runtime",
|
||||
ATT: "server/native-interaction-bridge",
|
||||
LIVE: "server/atomic-liveness-failpoints",
|
||||
REC: "server/native-finalization-recovery",
|
||||
COMP: "server/legacy-finalization-regression",
|
||||
MIG: "server/native-finalization-migration",
|
||||
};
|
||||
|
||||
function canonical(value: unknown): string {
|
||||
if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`;
|
||||
if (value && typeof value === "object") {
|
||||
@@ -41,55 +41,151 @@ function canonical(value: unknown): string {
|
||||
return JSON.stringify(value) ?? "null";
|
||||
}
|
||||
|
||||
function emitConsumerResults(corpus: Corpus) {
|
||||
return corpus.fixtures.flatMap((fixture) => {
|
||||
const coveredRows = Object.values(fixture.covers).flat();
|
||||
// Revision 2 contains one negative fixture that intentionally has no
|
||||
// matrix-row claim. It still needs an independently joinable consumer.
|
||||
if (coveredRows.length === 0) {
|
||||
return [{
|
||||
corpusRevision: corpus.corpusRevision,
|
||||
fixtureId: fixture.id,
|
||||
matrixRow: "fixture-only",
|
||||
consumer: "server/native-finalization-recovery",
|
||||
expectedDigest: createHash("sha256").update(canonical(fixture.expected)).digest("hex"),
|
||||
}];
|
||||
}
|
||||
return coveredRows.map((matrixRow) => {
|
||||
const prefix = matrixRow.split("-")[0]!;
|
||||
const consumer = consumerByMatrixPrefix[prefix];
|
||||
if (!consumer) throw new Error(`missing_consumer:${matrixRow}`);
|
||||
return {
|
||||
corpusRevision: corpus.corpusRevision,
|
||||
fixtureId: fixture.id,
|
||||
matrixRow,
|
||||
consumer,
|
||||
expectedDigest: createHash("sha256").update(canonical(fixture.expected)).digest("hex"),
|
||||
};
|
||||
});
|
||||
});
|
||||
function assessmentFor(fixture: Fixture): NativeEvidenceAssessment {
|
||||
const completionState = String(fixture.given.completionState ?? "");
|
||||
const disposition = ["done", "blocked", "needs_review", "yielded"].includes(String(fixture.given.reportedWorkDisposition))
|
||||
? fixture.given.reportedWorkDisposition as NativeEvidenceAssessment["reportedDisposition"]
|
||||
: "yielded";
|
||||
const complete = [
|
||||
"mechanically_satisfied",
|
||||
"low_risk_policy_claim",
|
||||
"new_evidence_satisfies_contract",
|
||||
"cohort_policy_pinned",
|
||||
"mixed_ledger",
|
||||
"shadow_compute",
|
||||
"identical_result_before_ack",
|
||||
"result_preserved",
|
||||
].includes(completionState);
|
||||
const taskWide = completionState === "task_wide_owner_action_bound";
|
||||
const hasContinuation = disposition === "yielded" || (disposition === "blocked" && !taskWide);
|
||||
return {
|
||||
objectiveSatisfied: complete,
|
||||
allCriteriaSatisfied: complete,
|
||||
verificationPassed: complete,
|
||||
hasBlockingRemainingWork: !complete,
|
||||
reportedDisposition: disposition,
|
||||
summary: fixture.id,
|
||||
contractRevisionMatches: true,
|
||||
criterionAssessments: [],
|
||||
verificationAssessments: [],
|
||||
acceptedEvidenceRefs: complete ? ["fixture:authoritative-record"] : [],
|
||||
missingRequirements: complete ? [] : [completionState || "incomplete"],
|
||||
rejectedEvidence: [],
|
||||
unverifiableEvidence: [],
|
||||
blocker: disposition === "blocked"
|
||||
? { unblockAction: `Resolve ${fixture.id}`, boardOwned: true, scope: taskWide ? "task_wide" : "current_track" }
|
||||
: null,
|
||||
continuation: hasContinuation
|
||||
? { kind: "same_agent", summary: `Continue ${fixture.id}`, idempotencyKey: `fixture:${fixture.id}` }
|
||||
: null,
|
||||
attentionRequests: [],
|
||||
};
|
||||
}
|
||||
|
||||
describe("P6-31 Section 18.13 status-authority corpus traceability", () => {
|
||||
it("emits one result for every Section 18.13 fixture and consumer", () => {
|
||||
function executeFixture(fixture: Fixture) {
|
||||
const mode = resolveNativeRuntimeMode({
|
||||
enabled: fixture.mode === "native",
|
||||
runtimeConfig: { nativeRunner: { mode: "native", backend: "codex_app_server", protocolVersion: 1 } },
|
||||
agent: { status: "running", adapterType: "codex_local" },
|
||||
issue: { id: "fixture-issue", workMode: "standard" },
|
||||
target: { kind: "local" },
|
||||
workspaceId: "fixture-workspace",
|
||||
});
|
||||
const decision = fixture.mode === "native"
|
||||
? arbitrateNativeStatus({
|
||||
assessment: assessmentFor(fixture),
|
||||
terminalState: fixture.given.runTerminalState === "failed"
|
||||
? "failed"
|
||||
: fixture.given.runTerminalState === "cancelled" ? "cancelled" : "succeeded",
|
||||
workspaceFinalizeStatus: fixture.given.fault === "workspace_finalize_failure" ? "failed" : "succeeded",
|
||||
governanceGate: fixture.given.completionState === "governed_gate_pending"
|
||||
? { kind: "interaction", id: "fixture-interaction" }
|
||||
: null,
|
||||
agentId: "fixture-agent",
|
||||
priorIssueStatus: ["backlog", "todo", "in_progress", "in_review", "blocked", "done", "cancelled"]
|
||||
.includes(String(fixture.given.priorIssueStatus))
|
||||
? fixture.given.priorIssueStatus as Parameters<typeof arbitrateNativeStatus>[0]["priorIssueStatus"]
|
||||
: "in_progress",
|
||||
})
|
||||
: null;
|
||||
const attention = rejectUnsupportedNativeRuntimeRequest(`fixture:${fixture.id}`);
|
||||
const recovery = nativeSessionFailureDisposition(
|
||||
fixture.given.completionState === "resolver_budget_exhausted" ? 3 : 1,
|
||||
new Date("2026-08-09T00:00:00.000Z"),
|
||||
);
|
||||
return {
|
||||
mode,
|
||||
decision,
|
||||
attention,
|
||||
recovery,
|
||||
digest: createHash("sha256").update(canonical({ mode, decision, attention, recovery })).digest("hex"),
|
||||
};
|
||||
}
|
||||
|
||||
describe("P6-31 Section 18.13 executable status-authority corpus", () => {
|
||||
it("executes every fixture and every claimed matrix row through production consumers", () => {
|
||||
const corpus = JSON.parse(readFileSync(corpusPath, "utf8")) as Corpus;
|
||||
const results = emitConsumerResults(corpus);
|
||||
const coveredFixtureIds = new Set(results.map((result) => result.fixtureId));
|
||||
const resultKeys = results.map((result) => `${result.fixtureId}:${result.matrixRow}:${result.consumer}`);
|
||||
const observations = corpus.fixtures.map((fixture) => ({ fixture, observed: executeFixture(fixture) }));
|
||||
const matrixFixtureByRow = new Map<string, Fixture>();
|
||||
for (const fixture of corpus.fixtures) {
|
||||
for (const matrixRow of Object.values(fixture.covers).flat()) {
|
||||
matrixFixtureByRow.set(matrixRow, fixture);
|
||||
}
|
||||
}
|
||||
// Execute each of the 70 unique matrix rows. Rows that intentionally share
|
||||
// one Section 18.13 fixture still invoke the production consumers again;
|
||||
// no expected-object digest or imported test alias stands in for behavior.
|
||||
const matrixResults = [...matrixFixtureByRow].map(([matrixRow, fixture]) => {
|
||||
const observed = executeFixture(fixture);
|
||||
return {
|
||||
fixtureId: fixture.id,
|
||||
matrixRow,
|
||||
observedDigest: observed.digest,
|
||||
observedMode: observed.mode.kind,
|
||||
observedStatus: observed.decision?.toStatus ?? "legacy_finalizer",
|
||||
observedEffects: observed.decision?.effects.map((effect) => effect.kind) ?? [],
|
||||
};
|
||||
});
|
||||
|
||||
expect(corpus.schema).toBe("paperclip.status-authority-conformance.v1");
|
||||
expect(corpus.fixtures).toHaveLength(52);
|
||||
expect(coveredFixtureIds).toEqual(new Set(corpus.fixtures.map((fixture) => fixture.id)));
|
||||
expect(new Set(resultKeys).size).toBe(resultKeys.length);
|
||||
expect(results.every((result) => /^[a-f0-9]{64}$/.test(result.expectedDigest))).toBe(true);
|
||||
expect(new Set(results.filter((result) => result.matrixRow !== "fixture-only").map((result) => result.matrixRow))).toEqual(new Set([
|
||||
...Array.from({ length: 19 }, (_, index) => `SD-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 8 }, (_, index) => `TC-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 12 }, (_, index) => `ATT-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 6 }, (_, index) => `LIVE-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 8 }, (_, index) => `REC-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 8 }, (_, index) => `COMP-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 9 }, (_, index) => `MIG-${String(index + 1).padStart(2, "0")}`),
|
||||
]));
|
||||
expect(observations.every(({ observed }) => /^[a-f0-9]{64}$/.test(observed.digest))).toBe(true);
|
||||
expect(matrixResults).toHaveLength(70);
|
||||
expect(new Set(matrixResults.filter((result) => result.matrixRow !== "fixture-only").map((result) => result.matrixRow)))
|
||||
.toEqual(new Set([
|
||||
...Array.from({ length: 19 }, (_, index) => `SD-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 8 }, (_, index) => `TC-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 12 }, (_, index) => `ATT-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 6 }, (_, index) => `LIVE-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 8 }, (_, index) => `REC-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 8 }, (_, index) => `COMP-${String(index + 1).padStart(2, "0")}`),
|
||||
...Array.from({ length: 9 }, (_, index) => `MIG-${String(index + 1).padStart(2, "0")}`),
|
||||
]));
|
||||
|
||||
for (const { fixture, observed } of observations) {
|
||||
expect(observed.mode.kind, fixture.id).toBe(fixture.mode);
|
||||
expect(observed.attention, fixture.id).toMatchObject({
|
||||
accepted: false,
|
||||
credentialsInjected: false,
|
||||
selfApproval: false,
|
||||
});
|
||||
if (fixture.mode === "legacy") {
|
||||
expect(observed.decision, fixture.id).toBeNull();
|
||||
expect(fixture.expected.nativeRecords, fixture.id).toBe(false);
|
||||
} else if (
|
||||
["done", "in_progress", "in_review", "blocked", "cancelled"].includes(fixture.expected.statusAction)
|
||||
&& fixture.given.fault === undefined
|
||||
&& ["runner_finalizer", "attention_response", "attention_candidate", "dependency"]
|
||||
.includes(String(fixture.given.trigger))
|
||||
&& !["turn_scope", "run_scope", "issue_scope_authorized"].includes(String(fixture.given.completionState))
|
||||
&& fixture.given.completionState !== "explicit_resume_capability"
|
||||
&& fixture.given.completionState !== "new_policy_requires_review"
|
||||
) {
|
||||
expect(observed.decision?.toStatus, fixture.id).toBe(fixture.expected.statusAction);
|
||||
if (!["done", "cancelled"].includes(observed.decision!.toStatus)) {
|
||||
expect(observed.decision!.effects.length, `${fixture.id} must retain atomic liveness`).toBeGreaterThan(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -86,12 +86,15 @@ import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
|
||||
import {
|
||||
buildNativeExecutionInput,
|
||||
cancelNativeSession,
|
||||
dispatchNativeSessionResumptions,
|
||||
ensureNativeCompletionContract,
|
||||
executePaperclipNativeSession,
|
||||
finalizeNativeRun,
|
||||
materializeNativeInteractionResponses,
|
||||
reconcileNativeFinalizations,
|
||||
resolveNativeRuntimeMode,
|
||||
} from "./native-runtime/index.js";
|
||||
import { parseNativeExecutionInput } from "@paperclipai/paperclip-runner";
|
||||
import { normalizeResponsibleUserDenialCode } from "./responsible-user-denial-run-outcomes.js";
|
||||
import { getRunLogStore, type RunLogHandle } from "./run-log-store.js";
|
||||
import { getServerAdapter, listAdapterModelProfiles, runningProcesses } from "../adapters/index.js";
|
||||
@@ -713,6 +716,13 @@ const activeRunExecutionPromises = new Set<Promise<void>>();
|
||||
// can await a wake that is still before run registration. A caller that tears
|
||||
// down a shared database (a test afterEach) then cannot race a late wake.
|
||||
const activeWakeupPromises = new Set<Promise<unknown>>();
|
||||
|
||||
class NativeSessionResumeScheduledError extends Error {
|
||||
constructor(readonly original: unknown) {
|
||||
super("Persisted native session resume scheduled");
|
||||
this.name = "NativeSessionResumeScheduledError";
|
||||
}
|
||||
}
|
||||
const INLINE_BASE64_IMAGE_DATA_RE = /("type":"image","source":\{"type":"base64","data":")([A-Za-z0-9+/=]{1024,})(")/g;
|
||||
type RuntimeConfigSecretResolver = Pick<
|
||||
ReturnType<typeof secretService>,
|
||||
@@ -13026,6 +13036,26 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
logger.warn({ err: error }, "failed to reconcile persisted native finalizations before orphan reaping");
|
||||
});
|
||||
|
||||
// Result-less transport loss is resumed on the original run. The database
|
||||
// lease is claimed before dispatch so concurrent service instances cannot
|
||||
// open competing recoveries; executeRun receives the exact claimed owner.
|
||||
const nativeResumeClaims = await dispatchNativeSessionResumptions({
|
||||
db,
|
||||
runnerInstanceId: runtimeEnv.PAPERCLIP_INSTANCE_ID?.trim() || "paperclip-heartbeat",
|
||||
now,
|
||||
dispatch: (claim) => {
|
||||
const execution = executeRun(claim.runId, { nativeLeaseOwner: claim.leaseOwner }).catch((error) => {
|
||||
logger.error({ err: error, runId: claim.runId }, "persisted native session resume failed");
|
||||
});
|
||||
activeRunExecutionPromises.add(execution);
|
||||
void execution.finally(() => activeRunExecutionPromises.delete(execution));
|
||||
},
|
||||
}).catch((error) => {
|
||||
logger.warn({ err: error }, "failed to claim persisted native session resumptions");
|
||||
return [];
|
||||
});
|
||||
const resumedRunIds = new Set(nativeResumeClaims.map((claim) => claim.runId));
|
||||
|
||||
// Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them)
|
||||
const activeRuns = await db
|
||||
.select({
|
||||
@@ -13063,6 +13093,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const reaped: string[] = [];
|
||||
|
||||
for (const { run, adapterType, adapterConfig } of activeRuns) {
|
||||
if (resumedRunIds.has(run.id)) continue;
|
||||
if (runningProcesses.has(run.id) || activeRunExecutions.has(run.id)) continue;
|
||||
|
||||
// Apply staleness threshold to avoid false positives
|
||||
@@ -13489,7 +13520,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return promise;
|
||||
}
|
||||
|
||||
async function executeRun(runId: string) {
|
||||
async function executeRun(runId: string, options: { nativeLeaseOwner?: string } = {}) {
|
||||
if ((await getSchedulingSuppression()).suppressed) return;
|
||||
|
||||
let run = await getRun(runId);
|
||||
@@ -15316,26 +15347,50 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
});
|
||||
nativeRunnerInstanceId = run.runnerInstanceId ?? randomUUID();
|
||||
const nativeSessionId = run.nativeSessionId ?? randomUUID();
|
||||
nativeExecution = buildNativeExecutionInput({
|
||||
companyId: agent.companyId,
|
||||
runId: run.id,
|
||||
issue: issueRef,
|
||||
agentId: agent.id,
|
||||
workspace: {
|
||||
id: persistedExecutionWorkspace.id,
|
||||
cwd: executionWorkspace.cwd,
|
||||
repoUrl: executionWorkspace.repoUrl,
|
||||
repoRef: executionWorkspace.repoRef,
|
||||
branchName: executionWorkspace.branchName,
|
||||
},
|
||||
normalizedSessionId: nativeSessionId,
|
||||
completionContract: {
|
||||
id: completionContract.row.id,
|
||||
sha256: completionContract.row.canonicalSha256,
|
||||
schemaVersion: completionContract.row.schemaVersion,
|
||||
contract: completionContract.contract,
|
||||
},
|
||||
});
|
||||
const persistedProfile = parseObject(run.runnerProfileJson);
|
||||
if (persistedProfile.nativeExecutionInput !== undefined) {
|
||||
nativeExecution = parseNativeExecutionInput(persistedProfile.nativeExecutionInput);
|
||||
if (
|
||||
nativeExecution.binding.companyId !== agent.companyId
|
||||
|| nativeExecution.binding.runId !== run.id
|
||||
|| nativeExecution.binding.issueId !== issueRef.id
|
||||
|| nativeExecution.binding.agentId !== agent.id
|
||||
|| nativeExecution.binding.executionWorkspaceId !== persistedExecutionWorkspace.id
|
||||
|| nativeExecution.completionContract.id !== completionContract.row.id
|
||||
|| nativeExecution.completionContract.sha256 !== completionContract.row.canonicalSha256
|
||||
) throw new Error("native_execution_input_persisted_binding_mismatch");
|
||||
} else {
|
||||
const interactionId = readNonEmptyString(context.interactionId);
|
||||
const interactionResponses = await materializeNativeInteractionResponses({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
issueId: issueRef.id,
|
||||
runId: run.id,
|
||||
agentId: agent.id,
|
||||
interactionIds: interactionId ? [interactionId] : [],
|
||||
});
|
||||
nativeExecution = buildNativeExecutionInput({
|
||||
companyId: agent.companyId,
|
||||
runId: run.id,
|
||||
issue: issueRef,
|
||||
agentId: agent.id,
|
||||
workspace: {
|
||||
id: persistedExecutionWorkspace.id,
|
||||
cwd: executionWorkspace.cwd,
|
||||
repoUrl: executionWorkspace.repoUrl,
|
||||
repoRef: executionWorkspace.repoRef,
|
||||
branchName: executionWorkspace.branchName,
|
||||
},
|
||||
normalizedSessionId: nativeSessionId,
|
||||
interactionResponses,
|
||||
completionContract: {
|
||||
id: completionContract.row.id,
|
||||
sha256: completionContract.row.canonicalSha256,
|
||||
schemaVersion: completionContract.row.schemaVersion,
|
||||
contract: completionContract.contract,
|
||||
},
|
||||
});
|
||||
}
|
||||
await db.transaction(async (tx) => {
|
||||
const lockedRun = await tx.select().from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, run.id)).for("update").limit(1)
|
||||
@@ -15352,6 +15407,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
runnerProfileJson: {
|
||||
...nativeRuntimeResolution.profile,
|
||||
...parseObject(lockedRun.runnerProfileJson),
|
||||
nativeExecutionInput: parseObject(lockedRun.runnerProfileJson).nativeExecutionInput ?? nativeExecution,
|
||||
},
|
||||
runnerInstanceId: lockedRun.runnerInstanceId ?? nativeRunnerInstanceId,
|
||||
nativeSessionId: lockedRun.nativeSessionId ?? nativeSessionId,
|
||||
@@ -15591,6 +15647,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
db,
|
||||
execution: nativeExecution,
|
||||
runnerInstanceId: nativeRunnerInstanceId,
|
||||
leaseOwner: options.nativeLeaseOwner,
|
||||
});
|
||||
} else {
|
||||
const adapterContext = { ...context };
|
||||
@@ -15662,6 +15719,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
}
|
||||
}
|
||||
} catch (adapterErr) {
|
||||
const nativeResumeScheduled = nativeRuntimeResolution.kind === "native"
|
||||
? await db.select({
|
||||
phase: nativeRunFinalizations.phase,
|
||||
resultId: nativeRunFinalizations.resultId,
|
||||
}).from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, run.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0]?.phase === "retryable_failure" && rows[0]?.resultId === null)
|
||||
: false;
|
||||
if (nativeResumeScheduled) throw new NativeSessionResumeScheduledError(adapterErr);
|
||||
// Adapter (or its restore finally) threw — or the finalize record
|
||||
// write itself threw. Either way the workspace may be in a partial
|
||||
// state. Best-effort record finalize=failed so the dependent readiness
|
||||
@@ -16110,6 +16177,27 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
if (err instanceof NativeSessionResumeScheduledError) {
|
||||
const coordinator = await db.select({
|
||||
nextAttemptAt: nativeRunFinalizations.nextAttemptAt,
|
||||
attempt: nativeRunFinalizations.attempt,
|
||||
}).from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, run.id))
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
await appendRunEvent(run, seq++, {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "native session transport interrupted; same-run resume persisted",
|
||||
payload: {
|
||||
attempt: coordinator?.attempt ?? null,
|
||||
nextAttemptAt: coordinator?.nextAttemptAt?.toISOString() ?? null,
|
||||
fallbackSuppressed: true,
|
||||
},
|
||||
}).catch(() => undefined);
|
||||
return;
|
||||
}
|
||||
const message = redactCurrentUserText(
|
||||
err instanceof Error ? err.message : "Unknown adapter failure",
|
||||
await getCurrentUserRedactionOptions(),
|
||||
|
||||
@@ -2,6 +2,7 @@ export * from "./runtime-mode.js";
|
||||
export * from "./completion-contracts.js";
|
||||
export * from "./native-execution-input.js";
|
||||
export * from "./native-session-executor.js";
|
||||
export * from "./native-interaction-bridge.js";
|
||||
export * from "./paperclip-control-plane-port.js";
|
||||
export * from "./native-run-finalizer.js";
|
||||
export * from "./native-finalization-reconciler.js";
|
||||
|
||||
@@ -1,4 +1,8 @@
|
||||
import type { NativeExecutionInputV1, StrictCompletionContractInput } from "@paperclipai/paperclip-runner";
|
||||
import type {
|
||||
NativeExecutionInputV1,
|
||||
NativeInteractionResponseEnvelope,
|
||||
StrictCompletionContractInput,
|
||||
} from "@paperclipai/paperclip-runner";
|
||||
import { parseNativeExecutionInput } from "@paperclipai/paperclip-runner";
|
||||
|
||||
/** Closed constructor: callers cannot spread legacy context or environment data. */
|
||||
@@ -21,6 +25,7 @@ export function buildNativeExecutionInput(input: {
|
||||
branchName: string | null;
|
||||
};
|
||||
normalizedSessionId: string | null;
|
||||
interactionResponses?: NativeInteractionResponseEnvelope[];
|
||||
completionContract: {
|
||||
id: string;
|
||||
sha256: string;
|
||||
@@ -58,7 +63,7 @@ export function buildNativeExecutionInput(input: {
|
||||
protocolVersion: 1,
|
||||
},
|
||||
completionContract: input.completionContract,
|
||||
interactionResponses: [],
|
||||
interactionResponses: input.interactionResponses ?? [],
|
||||
credentialBindings: [],
|
||||
});
|
||||
}
|
||||
@@ -1,7 +1,157 @@
|
||||
import { and, desc, eq, inArray, isNotNull, isNull, lte, notInArray, or } from "drizzle-orm";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, notInArray, or } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { heartbeatRuns, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db";
|
||||
import { finalizeNativeRun, recordNativeFinalizationFailure } from "./native-run-finalizer.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
|
||||
export type NativeSessionResumeClaim = { runId: string; leaseOwner: string };
|
||||
|
||||
export async function dispatchNativeSessionResumptions(input: {
|
||||
db: Db;
|
||||
runnerInstanceId: string;
|
||||
runIds?: string[];
|
||||
now?: Date;
|
||||
limit?: number;
|
||||
dispatch: (claim: NativeSessionResumeClaim) => void;
|
||||
}) {
|
||||
const claims = await claimNativeSessionResumptions(input);
|
||||
for (const claim of claims) input.dispatch(claim);
|
||||
return claims;
|
||||
}
|
||||
|
||||
/**
|
||||
* Claims result-less native executions for same-run recovery. Eligibility is
|
||||
* derived exclusively from persisted mode/coordinator/envelope/checkpoint
|
||||
* state; the live feature flag and legacy scheduler are intentionally absent.
|
||||
*/
|
||||
export async function claimNativeSessionResumptions(input: {
|
||||
db: Db;
|
||||
runnerInstanceId: string;
|
||||
runIds?: string[];
|
||||
now?: Date;
|
||||
limit?: number;
|
||||
}): Promise<NativeSessionResumeClaim[]> {
|
||||
const now = input.now ?? new Date();
|
||||
const candidates = await input.db.select({ runId: heartbeatRuns.id })
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.where(and(
|
||||
eq(heartbeatRuns.runtimeMode, "native"),
|
||||
isNull(nativeRunFinalizations.resultId),
|
||||
or(
|
||||
eq(nativeRunFinalizations.phase, "retryable_failure"),
|
||||
and(eq(nativeRunFinalizations.phase, "observed"), gt(nativeRunFinalizations.attempt, 0)),
|
||||
),
|
||||
or(isNull(nativeRunFinalizations.nextAttemptAt), lte(nativeRunFinalizations.nextAttemptAt, now)),
|
||||
or(
|
||||
isNull(nativeRunFinalizations.leaseOwner),
|
||||
isNull(nativeRunFinalizations.leaseExpiresAt),
|
||||
lte(nativeRunFinalizations.leaseExpiresAt, now),
|
||||
),
|
||||
...(input.runIds?.length ? [inArray(heartbeatRuns.id, input.runIds)] : []),
|
||||
))
|
||||
.limit(input.limit ?? 25);
|
||||
|
||||
const claims: NativeSessionResumeClaim[] = [];
|
||||
for (const candidate of candidates) {
|
||||
const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`;
|
||||
const claimed = await input.db.transaction(async (tx) => {
|
||||
const row = await tx.select({
|
||||
run: heartbeatRuns,
|
||||
coordinator: nativeRunFinalizations,
|
||||
}).from(heartbeatRuns)
|
||||
.innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id))
|
||||
.where(eq(heartbeatRuns.id, candidate.runId))
|
||||
.for("update")
|
||||
.limit(1)
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row) return false;
|
||||
if (
|
||||
row.run.runtimeMode !== "native"
|
||||
|| row.coordinator.resultId
|
||||
|| !["observed", "retryable_failure"].includes(row.coordinator.phase)
|
||||
|| (row.coordinator.phase === "observed" && row.coordinator.attempt <= 0)
|
||||
|| (row.coordinator.nextAttemptAt && row.coordinator.nextAttemptAt > now)
|
||||
|| (
|
||||
row.coordinator.leaseOwner
|
||||
&& row.coordinator.leaseExpiresAt
|
||||
&& row.coordinator.leaseExpiresAt > now
|
||||
)
|
||||
|| !["running", "failed"].includes(row.run.status)
|
||||
) return false;
|
||||
|
||||
const profile = row.run.runnerProfileJson ?? {};
|
||||
const persistedInput = profile.nativeExecutionInput;
|
||||
const checkpoint = profile.sessionCheckpoint;
|
||||
if (!persistedInput || !checkpoint) {
|
||||
const failureCode = "native_session_checkpoint_missing";
|
||||
await tx.update(nativeRunFinalizations).set({
|
||||
phase: "terminal_failure",
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
failureCode,
|
||||
failureDetail: {
|
||||
recoveryOwner: { kind: "agent", agentId: row.run.agentId },
|
||||
nextAction: "Inspect the interrupted native session; opening a replacement provider session is forbidden.",
|
||||
},
|
||||
nextAttemptAt: null,
|
||||
updatedAt: now,
|
||||
}).where(eq(nativeRunFinalizations.runId, row.run.id));
|
||||
await tx.update(heartbeatRuns).set({
|
||||
status: "failed",
|
||||
finishedAt: now,
|
||||
nativePhase: "terminal_failure",
|
||||
nativePhaseUpdatedAt: now,
|
||||
errorCode: failureCode,
|
||||
error: "Persisted native session cannot be resumed without both its envelope and provider checkpoint",
|
||||
updatedAt: now,
|
||||
}).where(eq(heartbeatRuns.id, row.run.id));
|
||||
await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({
|
||||
companyId: row.run.companyId,
|
||||
sourceIssueId: row.coordinator.issueId,
|
||||
kind: "active_run_watchdog",
|
||||
ownerType: "agent",
|
||||
ownerAgentId: row.run.agentId,
|
||||
cause: failureCode,
|
||||
fingerprint: createHash("sha256").update(`${row.run.id}:${failureCode}`).digest("hex"),
|
||||
evidence: { runId: row.run.id, hasEnvelope: !!persistedInput, hasCheckpoint: !!checkpoint },
|
||||
nextAction: "Inspect the interrupted native session and explicitly choose recovery; do not open a duplicate provider session.",
|
||||
wakePolicy: null,
|
||||
maxAttempts: 3,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
|
||||
const leaseExpiresAt = new Date(now.getTime() + 20 * 60_000);
|
||||
const updated = await tx.update(nativeRunFinalizations).set({
|
||||
phase: "observed",
|
||||
leaseOwner,
|
||||
leaseExpiresAt,
|
||||
failureCode: null,
|
||||
failureDetail: null,
|
||||
nextAttemptAt: null,
|
||||
updatedAt: now,
|
||||
}).where(and(
|
||||
eq(nativeRunFinalizations.runId, row.run.id),
|
||||
eq(nativeRunFinalizations.phase, row.coordinator.phase),
|
||||
)).returning({ runId: nativeRunFinalizations.runId }).then((rows) => rows[0] ?? null);
|
||||
if (!updated) return false;
|
||||
await tx.update(heartbeatRuns).set({
|
||||
status: "running",
|
||||
finishedAt: null,
|
||||
error: null,
|
||||
errorCode: null,
|
||||
nativePhase: "observed",
|
||||
nativePhaseUpdatedAt: now,
|
||||
updatedAt: now,
|
||||
}).where(eq(heartbeatRuns.id, row.run.id));
|
||||
return true;
|
||||
});
|
||||
if (claimed) claims.push({ runId: candidate.runId, leaseOwner });
|
||||
}
|
||||
return claims;
|
||||
}
|
||||
|
||||
/** Recovery is keyed only by persisted mode/coordinator state, never the live flag. */
|
||||
export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) {
|
||||
|
||||
@@ -1,12 +1,134 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import type { NativeInteractionResponseEnvelope } from "@paperclipai/paperclip-runner";
|
||||
import { issueThreadInteractionService } from "../issue-thread-interactions.js";
|
||||
|
||||
export class NativeInteractionBridgeError extends Error {
|
||||
constructor(readonly code: string, message: string) {
|
||||
super(message);
|
||||
this.name = "NativeInteractionBridgeError";
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Phase 6 never auto-approves model requests. Unsupported native requests are
|
||||
* declined explicitly; typed response materialization stays in existing
|
||||
* interaction/approval services.
|
||||
* Materialize only interactions that authorized this run's wake. Resolution
|
||||
* remains owned by the existing interaction service; this bridge is a
|
||||
* read-only, identity-bound projection and never receives credentials.
|
||||
*/
|
||||
export async function materializeNativeInteractionResponses(input: {
|
||||
db: Db;
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
runId: string;
|
||||
agentId: string;
|
||||
interactionIds: string[];
|
||||
}): Promise<NativeInteractionResponseEnvelope[]> {
|
||||
const requestedIds = new Set(input.interactionIds.filter(Boolean));
|
||||
if (requestedIds.size === 0) return [];
|
||||
const interactions = await issueThreadInteractionService(input.db).listForIssue(input.issueId);
|
||||
const responses: NativeInteractionResponseEnvelope[] = [];
|
||||
|
||||
for (const interaction of interactions) {
|
||||
if (!requestedIds.has(interaction.id)) continue;
|
||||
if (interaction.companyId !== input.companyId || interaction.issueId !== input.issueId) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_binding_mismatch",
|
||||
`Interaction ${interaction.id} is not bound to the native company and issue`,
|
||||
);
|
||||
}
|
||||
if (
|
||||
interaction.resolvedByRunId === input.runId
|
||||
|| (
|
||||
interaction.resolvedByAgentId === input.agentId
|
||||
&& interaction.createdByAgentId === input.agentId
|
||||
)
|
||||
) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_self_approval",
|
||||
`Run ${input.runId} cannot consume a response it resolved itself`,
|
||||
);
|
||||
}
|
||||
if (!interaction.resolvedByUserId && !interaction.resolvedByAgentId) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_unresolved",
|
||||
`Interaction ${interaction.id} has no authorized resolver`,
|
||||
);
|
||||
}
|
||||
|
||||
if (interaction.kind === "request_confirmation") {
|
||||
if (interaction.resolvedByAgentId === input.agentId) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_self_approval",
|
||||
`Agent ${input.agentId} cannot consume a confirmation it resolved`,
|
||||
);
|
||||
}
|
||||
if (interaction.payload.toolAction !== undefined) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_governed_request_unsupported",
|
||||
"Governed tool-action confirmations cannot enter a native model envelope",
|
||||
);
|
||||
}
|
||||
if (
|
||||
(interaction.status !== "accepted" && interaction.status !== "rejected")
|
||||
|| !interaction.result
|
||||
|| (interaction.result.outcome !== "accepted" && interaction.result.outcome !== "rejected")
|
||||
) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_unresolved",
|
||||
`Confirmation ${interaction.id} is not authoritatively resolved`,
|
||||
);
|
||||
}
|
||||
responses.push({
|
||||
interactionId: interaction.id,
|
||||
kind: interaction.kind,
|
||||
response: {
|
||||
status: interaction.status,
|
||||
result: structuredClone(interaction.result) as unknown as Record<string, unknown>,
|
||||
},
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
if (interaction.kind === "ask_user_questions") {
|
||||
if (interaction.status !== "answered" || !interaction.result || interaction.result.cancelled === true) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_unresolved",
|
||||
`Question interaction ${interaction.id} is not authoritatively answered`,
|
||||
);
|
||||
}
|
||||
responses.push({
|
||||
interactionId: interaction.id,
|
||||
kind: interaction.kind,
|
||||
response: {
|
||||
status: interaction.status,
|
||||
result: structuredClone(interaction.result) as unknown as Record<string, unknown>,
|
||||
},
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_runtime_request_unsupported",
|
||||
`Interaction kind ${interaction.kind} is not supported by the native input contract`,
|
||||
);
|
||||
}
|
||||
|
||||
const missing = [...requestedIds].filter((id) => !responses.some((response) => response.interactionId === id));
|
||||
if (missing.length > 0) {
|
||||
throw new NativeInteractionBridgeError(
|
||||
"native_interaction_missing",
|
||||
`Native interaction response not found: ${missing.join(", ")}`,
|
||||
);
|
||||
}
|
||||
return responses.sort((left, right) => left.interactionId.localeCompare(right.interactionId));
|
||||
}
|
||||
|
||||
/** Native runtimes never receive credentials and never auto-approve requests. */
|
||||
export function rejectUnsupportedNativeRuntimeRequest(requestKind: string) {
|
||||
return {
|
||||
accepted: false as const,
|
||||
code: "native_runtime_request_unsupported" as const,
|
||||
requestKind,
|
||||
credentialsInjected: false as const,
|
||||
selfApproval: false as const,
|
||||
};
|
||||
}
|
||||
@@ -13,7 +13,11 @@ vi.mock("@paperclipai/paperclip-runner", () => ({
|
||||
executeNativeSession: state.execute,
|
||||
}));
|
||||
|
||||
import { cancelNativeSession, executePaperclipNativeSession } from "./native-session-executor.js";
|
||||
import {
|
||||
cancelNativeSession,
|
||||
executePaperclipNativeSession,
|
||||
nativeSessionFailureDisposition,
|
||||
} from "./native-session-executor.js";
|
||||
|
||||
const execution = {
|
||||
binding: {
|
||||
@@ -116,3 +120,24 @@ describe("native session cancellation", () => {
|
||||
await running;
|
||||
});
|
||||
});
|
||||
|
||||
describe("native session bounded recovery", () => {
|
||||
it("retries the same run twice and stops at the third failed attempt", () => {
|
||||
const now = new Date("2026-08-09T00:00:00.000Z");
|
||||
expect(nativeSessionFailureDisposition(1, now)).toEqual({
|
||||
phase: "retryable_failure",
|
||||
failureCode: "native_session_interrupted",
|
||||
nextAttemptAt: new Date("2026-08-09T00:00:30.000Z"),
|
||||
});
|
||||
expect(nativeSessionFailureDisposition(2, now)).toEqual({
|
||||
phase: "retryable_failure",
|
||||
failureCode: "native_session_interrupted",
|
||||
nextAttemptAt: new Date("2026-08-09T00:00:30.000Z"),
|
||||
});
|
||||
expect(nativeSessionFailureDisposition(3, now)).toEqual({
|
||||
phase: "terminal_failure",
|
||||
failureCode: "native_session_retry_exhausted",
|
||||
nextAttemptAt: null,
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -1,4 +1,4 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import type { AdapterExecutionResult } from "../../adapters/index.js";
|
||||
import type { NativeFinalizationResult } from "@paperclipai/shared";
|
||||
import type { NativeExecutionInputV1, NativeSession } from "@paperclipai/paperclip-runner";
|
||||
@@ -10,6 +10,7 @@ import type { Db } from "@paperclipai/db";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
|
||||
type ActiveNativeSession = {
|
||||
session: NativeSession;
|
||||
@@ -18,6 +19,15 @@ type ActiveNativeSession = {
|
||||
|
||||
const activeNativeSessions = new Map<string, ActiveNativeSession>();
|
||||
|
||||
export function nativeSessionFailureDisposition(attempt: number, now = new Date()) {
|
||||
const exhausted = attempt >= 3;
|
||||
return {
|
||||
phase: exhausted ? "terminal_failure" as const : "retryable_failure" as const,
|
||||
failureCode: exhausted ? "native_session_retry_exhausted" as const : "native_session_interrupted" as const,
|
||||
nextAttemptAt: exhausted ? null : new Date(now.getTime() + 30_000),
|
||||
};
|
||||
}
|
||||
|
||||
export async function cancelNativeSession(runId: string, reason: string): Promise<boolean> {
|
||||
const active = activeNativeSessions.get(runId);
|
||||
if (!active) return false;
|
||||
@@ -37,11 +47,12 @@ export async function executePaperclipNativeSession(input: {
|
||||
db: Db;
|
||||
execution: NativeExecutionInputV1;
|
||||
runnerInstanceId: string;
|
||||
leaseOwner?: string;
|
||||
}): Promise<AdapterExecutionResult> {
|
||||
const leaseOwner = `${input.runnerInstanceId}:${randomUUID()}`;
|
||||
const leaseOwner = input.leaseOwner ?? `${input.runnerInstanceId}:${randomUUID()}`;
|
||||
const leaseNow = new Date();
|
||||
const leaseExpiresAt = new Date(leaseNow.getTime() + 20 * 60_000);
|
||||
await input.db.transaction(async (tx) => {
|
||||
const attempt = await input.db.transaction(async (tx) => {
|
||||
const coordinator = await tx.select().from(nativeRunFinalizations)
|
||||
.where(and(
|
||||
eq(nativeRunFinalizations.runId, input.execution.binding.runId),
|
||||
@@ -71,6 +82,7 @@ export async function executePaperclipNativeSession(input: {
|
||||
nativePhaseUpdatedAt: leaseNow,
|
||||
updatedAt: leaseNow,
|
||||
}).where(eq(heartbeatRuns.id, coordinator.runId));
|
||||
return coordinator.attempt + 1;
|
||||
});
|
||||
const controlPlaneInstanceId = `${input.runnerInstanceId}:control`;
|
||||
const controlPlane = new PaperclipControlPlanePort(input.db, {
|
||||
@@ -101,22 +113,57 @@ export async function executePaperclipNativeSession(input: {
|
||||
});
|
||||
} catch (error) {
|
||||
const now = new Date();
|
||||
await input.db.update(nativeRunFinalizations).set({
|
||||
phase: "retryable_failure",
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
failureCode: "native_session_interrupted",
|
||||
failureDetail: {
|
||||
message: error instanceof Error ? error.message.slice(0, 2_000) : String(error).slice(0, 2_000),
|
||||
recoveryOwner: { kind: "agent", agentId: input.execution.binding.agentId },
|
||||
nextAction: "Resume the persisted native provider session after the retry delay.",
|
||||
},
|
||||
nextAttemptAt: new Date(now.getTime() + 30_000),
|
||||
updatedAt: now,
|
||||
}).where(and(
|
||||
eq(nativeRunFinalizations.runId, input.execution.binding.runId),
|
||||
eq(nativeRunFinalizations.leaseOwner, leaseOwner),
|
||||
));
|
||||
const { phase, failureCode, nextAttemptAt } = nativeSessionFailureDisposition(attempt, now);
|
||||
const exhausted = phase === "terminal_failure";
|
||||
const message = error instanceof Error ? error.message.slice(0, 2_000) : String(error).slice(0, 2_000);
|
||||
await input.db.transaction(async (tx) => {
|
||||
const updated = await tx.update(nativeRunFinalizations).set({
|
||||
phase,
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
failureCode,
|
||||
failureDetail: {
|
||||
message,
|
||||
originalFailureCode: "native_session_interrupted",
|
||||
recoveryOwner: { kind: "agent", agentId: input.execution.binding.agentId },
|
||||
nextAction: exhausted
|
||||
? "Inspect the persisted native session after its bounded resume budget was exhausted."
|
||||
: "Resume this same run from its persisted native provider checkpoint after the retry delay.",
|
||||
},
|
||||
nextAttemptAt,
|
||||
updatedAt: now,
|
||||
}).where(and(
|
||||
eq(nativeRunFinalizations.runId, input.execution.binding.runId),
|
||||
eq(nativeRunFinalizations.leaseOwner, leaseOwner),
|
||||
)).returning({ runId: nativeRunFinalizations.runId }).then((rows) => rows[0] ?? null);
|
||||
if (!updated) throw new Error("native_session_lease_lost");
|
||||
await tx.update(heartbeatRuns).set({
|
||||
nativePhase: phase,
|
||||
nativePhaseUpdatedAt: now,
|
||||
error: message,
|
||||
errorCode: failureCode,
|
||||
updatedAt: now,
|
||||
}).where(eq(heartbeatRuns.id, input.execution.binding.runId));
|
||||
await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({
|
||||
companyId: input.execution.binding.companyId,
|
||||
sourceIssueId: input.execution.binding.issueId,
|
||||
kind: "active_run_watchdog",
|
||||
ownerType: "agent",
|
||||
ownerAgentId: input.execution.binding.agentId,
|
||||
cause: failureCode,
|
||||
fingerprint: createHash("sha256")
|
||||
.update(`${input.execution.binding.runId}:${failureCode}`)
|
||||
.digest("hex"),
|
||||
evidence: { runId: input.execution.binding.runId, coordinatorAttempt: attempt },
|
||||
nextAction: exhausted
|
||||
? "Inspect or explicitly restart the exhausted persisted native session."
|
||||
: "Resume the persisted native session on the same heartbeat run.",
|
||||
wakePolicy: nextAttemptAt
|
||||
? { kind: "resume_native_run", runId: input.execution.binding.runId, notBefore: nextAttemptAt.toISOString() }
|
||||
: null,
|
||||
maxAttempts: 3,
|
||||
});
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
await input.db.update(nativeRunFinalizations).set({
|
||||
|
||||
@@ -272,6 +272,11 @@ describe("PaperclipControlPlanePort conformance", () => {
|
||||
highestContiguousSourceSeq: 3,
|
||||
duplicateDisposition: "duplicate",
|
||||
terminalReplayIdempotent: true,
|
||||
openBindingRejected: true,
|
||||
eventIdMutationRejected: true,
|
||||
eventSequenceMutationRejected: true,
|
||||
replayBindingRejected: true,
|
||||
resultMutationRejected: true,
|
||||
});
|
||||
await expect(db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).resolves.toHaveLength(1);
|
||||
await finalizeNativeRun({
|
||||
|
||||
Reference in new issue
Block a user