mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
Merge remote-tracking branch 'origin/master' into feat/one-process-identity
* origin/master: fix(recovery): require a 1 s start-time gap and treat EPERM as alive in owner proofs (#19848) feat(observability): anchor native time to first token at the accepted wake (#19838) # Conflicts: # server/src/services/native-runtime/native-restart-recovery.ts
This commit is contained in:
commit
b46d071573
15 files changed
+550
-48
No files matched your search
+11
-4
@@ -746,10 +746,17 @@ accepted ingress → queue → task.prepare → Runner startup → provider work
|
||||
└────────────────────── task.time_to_first_token ──────────────────────────────────┘
|
||||
```
|
||||
|
||||
The span starts at the native run trace's historical start:
|
||||
the accepted comment or attested question response when present, otherwise run
|
||||
creation from the queue span. Callers without preparation history start at native
|
||||
dispatch. The interval includes queueing, Core preparation, workspace preparation,
|
||||
The span starts at the earliest accepted input. The accepted comment or attested
|
||||
question response is the start only when its span is the earliest preparation span.
|
||||
A comment coalesced into an already-queued run starts after run creation, so that
|
||||
run starts at run creation from the queue span and reports `run_created`. The
|
||||
durable wake request's acceptance time replaces the start when it is earlier
|
||||
(deferred, coalesced and retried wakes). The `anchor` attribute (`comment`, `question_response`,
|
||||
`wake_request`, `run_created`) names which one. The wake request time is read
|
||||
after the first text in detached diagnostic work bounded to 2 s (PostgreSQL
|
||||
cancels a stalled read through a transaction-local `statement_timeout`), so it never adds a
|
||||
wait before `turn.start` and never delays event delivery. Callers without
|
||||
preparation history start at native dispatch and report `run_created`. The interval includes queueing, Core preparation, workspace preparation,
|
||||
Runner startup, provider latency, transport, and durable event ingestion. It ends
|
||||
in Core's commit callback, when the text is readable through the run-events API,
|
||||
using the Core clock rather than the provider event timestamp. Browser delivery
|
||||
|
||||
+13
-4
@@ -68,10 +68,19 @@ instance run log and add no first-party Telemetry or OpenTelemetry export.
|
||||
dispatch at the first newly committed deliverable text: assistant or reasoning
|
||||
`item.delta`/`item.completed`, a redaction tail released at completion, or an
|
||||
accepted run-result summary. Its
|
||||
`durationMs` starts at the native run trace's historical start (accepted ingress,
|
||||
otherwise queued run creation, or dispatch when history is absent) and ends at Core
|
||||
commit callback. `startOffsetMs` is zero, and `eventKind` is `agentMessage` or
|
||||
`reasoning`. It includes preparation and startup, but excludes browser
|
||||
`durationMs` starts at the earliest accepted input. The accepted comment or
|
||||
answered question is the start only when its span is the earliest preparation span.
|
||||
A comment coalesced into an already-queued run starts after run creation, so that
|
||||
run uses queued run creation. The durable wake request's `requested_at` replaces
|
||||
the start when it is earlier (deferred, coalesced and retried wakes). Without
|
||||
preparation history, the start is dispatch and the anchor is `run_created`. It ends at Core commit callback. `anchor` names the start:
|
||||
`comment`, `question_response`, `wake_request` or `run_created`. `startOffsetMs` is
|
||||
zero, or negative when a wake request precedes the run trace. `eventKind` is
|
||||
`agentMessage` or `reasoning`. The wake request time is read after the first text,
|
||||
in detached diagnostic work bounded to 2 s (a
|
||||
transaction-local `statement_timeout`, so PostgreSQL cancels a stalled read), so the measurement adds no wait before
|
||||
`turn.start` and never delays event delivery. The span can appear after later
|
||||
run-log rows; the trace settles it before `task.provider_session.measured`. It includes preparation and startup, but excludes browser
|
||||
delivery/rendering. This measures text availability, not an exact model token.
|
||||
Empty text, item starts, tools, usage, result proposals, and duplicate replay do
|
||||
not qualify. No deliverable text means no timing sample. A recovered dispatch can
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { once } from "node:events";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { processExists, processStartsDefinitelyDiffer } from "./controller-process-identity.js";
|
||||
import { isPidAlive, isProcessGroupAlive } from "./local-service-supervisor.js";
|
||||
import { readProcessStartedAt } from "./hot-restart.js";
|
||||
import { verifiedRecoveryProcessIsAlive } from "./native-runtime/native-session-executor.js";
|
||||
|
||||
describe("shared process identity proof", () => {
|
||||
afterEach(() => vi.restoreAllMocks());
|
||||
|
||||
it("needs a start-time gap of at least one second to prove a recycled PID", () => {
|
||||
const at = new Date("2026-11-07T10:00:00.000Z");
|
||||
expect(processStartsDefinitelyDiffer(at, new Date(at.getTime() + 999))).toBe(false);
|
||||
expect(processStartsDefinitelyDiffer(at, "2026-11-07T09:59:59.001Z")).toBe(false);
|
||||
expect(processStartsDefinitelyDiffer(at, new Date(at.getTime() + 1_000))).toBe(true);
|
||||
expect(processStartsDefinitelyDiffer(at, null)).toBe(false);
|
||||
expect(processStartsDefinitelyDiffer("not a date", at)).toBe(false);
|
||||
});
|
||||
|
||||
it("treats EPERM as a live process in every signal-0 check", () => {
|
||||
vi.spyOn(process, "kill").mockImplementation(() => {
|
||||
throw Object.assign(new Error("operation not permitted"), { code: "EPERM" });
|
||||
});
|
||||
expect(processExists(4242)).toBe(true);
|
||||
expect(isPidAlive(4242)).toBe(true);
|
||||
if (process.platform !== "win32" && process.platform !== "linux") expect(isProcessGroupAlive(4242)).toBe(true);
|
||||
});
|
||||
|
||||
it("treats ESRCH as a gone process", () => {
|
||||
vi.spyOn(process, "kill").mockImplementation(() => {
|
||||
throw Object.assign(new Error("no such process"), { code: "ESRCH" });
|
||||
});
|
||||
expect(processExists(4242)).toBe(false);
|
||||
expect(isPidAlive(4242)).toBe(false);
|
||||
});
|
||||
|
||||
it.skipIf(process.platform === "win32")("keeps a live recovered runner whose recorded start differs by under a second", async () => {
|
||||
const live = spawn(process.execPath, ["-e", "setInterval(() => {}, 1_000)"], { stdio: "ignore" });
|
||||
await once(live, "spawn");
|
||||
try {
|
||||
const observed = new Date((await readProcessStartedAt(live.pid!))!);
|
||||
const startedAt = new Date(observed.getTime() + 500).toISOString();
|
||||
await expect(verifiedRecoveryProcessIsAlive({ pid: live.pid!, startedAt })).resolves.toBe(true);
|
||||
const recycled = new Date(observed.getTime() + 5_000).toISOString();
|
||||
await expect(verifiedRecoveryProcessIsAlive({ pid: live.pid!, startedAt: recycled })).resolves.toBe(false);
|
||||
} finally {
|
||||
const exited = once(live, "exit");
|
||||
live.kill("SIGKILL");
|
||||
await exited;
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -106,7 +106,11 @@ export function readControllerProcess(value: unknown): ControllerProcess | null
|
||||
processStartedAt: typeof record.processStartedAt === "string" ? record.processStartedAt : null };
|
||||
}
|
||||
|
||||
function processExists(pid: number): boolean {
|
||||
/**
|
||||
* True when some process owns the PID in this PID space. EPERM means the
|
||||
* process exists but belongs to another user, so it counts as alive.
|
||||
*/
|
||||
export function processExists(pid: number): boolean {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
@@ -115,6 +119,24 @@ function processExists(pid: number): boolean {
|
||||
}
|
||||
}
|
||||
|
||||
/** Process start times closer than this may be the same process. */
|
||||
export const PROCESS_START_PRECISION_MS = 1_000;
|
||||
|
||||
/**
|
||||
* True only when two start times for one PID prove different processes.
|
||||
* `ps -o lstart` on macOS and BSD has whole-second precision, while a recorded
|
||||
* start can come from a millisecond clock, so a sub-second gap is ambiguous
|
||||
* and fails closed (same process). A missing or unparsable time proves nothing.
|
||||
*/
|
||||
export function processStartsDefinitelyDiffer(
|
||||
recorded: Date | string | null | undefined,
|
||||
observed: Date | string | null | undefined,
|
||||
): boolean {
|
||||
const left = recorded instanceof Date ? recorded.getTime() : Date.parse(recorded ?? "");
|
||||
const right = observed instanceof Date ? observed.getTime() : Date.parse(observed ?? "");
|
||||
return Number.isFinite(left) && Number.isFinite(right) && Math.abs(left - right) >= PROCESS_START_PRECISION_MS;
|
||||
}
|
||||
|
||||
/** True only with proof, in this process's own PID space, that the recorded
|
||||
* server process has exited. Another space, an unreadable space, this server,
|
||||
* or an ambiguous live PID returns false. */
|
||||
@@ -123,7 +145,6 @@ export async function controllerProcessExited(controller: ControllerProcess): Pr
|
||||
if (!processExists(controller.pid)) return true;
|
||||
// The PID now belongs to a live process, possibly this restarted server.
|
||||
// Only a definitely different start time proves the recorded process is gone.
|
||||
const recorded = Date.parse(controller.processStartedAt ?? "");
|
||||
const observed = await observedProcessStartedAt(controller.pid);
|
||||
return Number.isFinite(recorded) && observed !== null && Math.abs(observed - recorded) >= 1_000;
|
||||
return observed !== null && processStartsDefinitelyDiffer(controller.processStartedAt, new Date(observed));
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import path from "node:path";
|
||||
import { setTimeout as delay } from "node:timers/promises";
|
||||
import { promisify } from "node:util";
|
||||
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
import { processExists } from "./controller-process-identity.js";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
@@ -242,24 +243,16 @@ export async function findLocalServiceRegistryRecordByRuntimeServiceId(input: {
|
||||
return candidate;
|
||||
}
|
||||
|
||||
/** EPERM means the process exists under another user, so it is alive. */
|
||||
export function isPidAlive(pid: number) {
|
||||
if (!Number.isInteger(pid) || pid <= 0) return false;
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
return processExists(pid);
|
||||
}
|
||||
|
||||
export function isProcessGroupAlive(processGroupId: number | null | undefined) {
|
||||
if (process.platform === "win32") return false;
|
||||
if (typeof processGroupId !== "number" || !Number.isInteger(processGroupId) || processGroupId <= 0) return false;
|
||||
try {
|
||||
process.kill(-processGroupId, 0);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
if (!processExists(-processGroupId)) return false;
|
||||
|
||||
if (process.platform === "linux") {
|
||||
const liveMember = readLinuxProcessGroupActivity(processGroupId);
|
||||
|
||||
@@ -17,6 +17,7 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { agentExecutionsHaveStopped } from "../agent-execution-stop.js";
|
||||
import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import { environmentService } from "../environments.js";
|
||||
import { executionControlSchedule } from "../execution-control-work.js";
|
||||
@@ -253,6 +254,32 @@ describe("failed native run owner release", () => {
|
||||
}
|
||||
});
|
||||
|
||||
// `ps -o lstart` on macOS reports whole seconds, while a recorded start can
|
||||
// carry milliseconds. A sub-second gap must not be read as a recycled PID.
|
||||
it.each(["controller", "run process"])("keeps the owner when the live %s start time differs by under a second", async (holder) => {
|
||||
const { runId } = await nativeRun({ status: "failed", errorCode: "adapter_failed",
|
||||
coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } });
|
||||
const live = spawn(process.execPath, ["-e", "setInterval(() => {}, 1_000)"], { stdio: "ignore", detached: true });
|
||||
await once(live, "spawn");
|
||||
try {
|
||||
const observed = new Date((await readProcessStartedAt(live.pid!))!);
|
||||
const recorded = new Date(observed.getTime() + 500);
|
||||
if (holder === "controller") {
|
||||
await db.update(nativeRunFinalizations).set({ controllerPid: live.pid!, controllerBootId: randomUUID(),
|
||||
controllerProcessStartedAt: recorded }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
} else {
|
||||
await db.update(heartbeatRuns).set({ processPid: live.pid!, processGroupId: live.pid!, processStartedAt: recorded })
|
||||
.where(eq(heartbeatRuns.id, runId));
|
||||
}
|
||||
await expect(releaseAbandonedNativeOwners(db)).resolves.toBe(0);
|
||||
expect(await coordinator(runId)).toMatchObject({ phase: "observed", leaseOwner: expect.any(String) });
|
||||
} finally {
|
||||
const exited = once(live, "exit");
|
||||
live.kill("SIGKILL");
|
||||
await exited;
|
||||
}
|
||||
});
|
||||
|
||||
it("rotates bounded passes past owners that a live controller still holds", async () => {
|
||||
const runs = await Promise.all([0, 1, 2].map(() => nativeRun({ status: "failed", errorCode: "adapter_failed",
|
||||
coordinator: { leaseOwner: `default:resume:${randomUUID()}`, leaseExpiresAt: new Date(0) } })));
|
||||
|
||||
@@ -2,6 +2,7 @@ import { and, asc, eq, gt, inArray, isNotNull, isNull, lte, or, sql } from "driz
|
||||
import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { logger } from "../../middleware/logger.js";
|
||||
import { adapterExecutionControls } from "../adapter-execution-control.js";
|
||||
import { processExists, processStartsDefinitelyDiffer } from "../controller-process-identity.js";
|
||||
import { logActivity, publishActivity, type ActivityPublication } from "../activity-log.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
|
||||
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
@@ -85,8 +86,7 @@ export async function nextAbandonedNativeOwnerReleaseAt(db: Db): Promise<Date |
|
||||
}
|
||||
|
||||
function processAbsent(pid: number): boolean {
|
||||
try { process.kill(pid, 0); return false; }
|
||||
catch (error) { return (error as NodeJS.ErrnoException).code === "ESRCH"; }
|
||||
return !processExists(pid);
|
||||
}
|
||||
|
||||
/** True unless the recorded process is gone or provably a different process. */
|
||||
@@ -94,7 +94,7 @@ async function recordedProcessMayBeLive(pid: number | null, startedAt: Date | nu
|
||||
if (!pid || processAbsent(pid)) return false;
|
||||
if (!startedAt) return true;
|
||||
const observed = await readProcessStartedAt(pid).catch(() => null);
|
||||
return !observed || new Date(observed).getTime() === startedAt.getTime();
|
||||
return !processStartsDefinitelyDiffer(startedAt, observed);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -110,7 +110,7 @@ async function runProcessMayBeLive(run: Run, now: Date): Promise<boolean> {
|
||||
let groupAlive = Boolean(run.processGroupId && !processAbsent(-run.processGroupId));
|
||||
if (processAlive && run.processStartedAt) {
|
||||
const observed = await readProcessStartedAt(run.processPid!).catch(() => null);
|
||||
if (observed && new Date(observed).getTime() !== run.processStartedAt.getTime()) {
|
||||
if (processStartsDefinitelyDiffer(run.processStartedAt, observed)) {
|
||||
processAlive = false;
|
||||
if (run.processGroupId === run.processPid) groupAlive = false;
|
||||
}
|
||||
|
||||
@@ -35,7 +35,7 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import { REMOTE_CONTROLLER_LEASE_MS } from "./native-runner-recovery-contract.js";
|
||||
import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import { currentControllerBootId, currentProcessStartedAtMs } from "../controller-process-identity.js";
|
||||
import { currentControllerBootId, currentProcessStartedAtMs, processStartsDefinitelyDiffer } from "../controller-process-identity.js";
|
||||
import { getServerProcessStartedAt } from "../../server-info.js";
|
||||
import { redactSensitiveText } from "../../redaction.js";
|
||||
import { isUniqueViolation } from "../../db-errors.js";
|
||||
@@ -352,12 +352,8 @@ function definitivelyDifferentProcessStart(
|
||||
left: Date | null,
|
||||
right: Date | null,
|
||||
): boolean {
|
||||
if (left === null || right === null) return false;
|
||||
// `ps -o lstart` has only whole-second precision on macOS and BSD. A
|
||||
// sub-second disagreement can therefore be the same process when one probe
|
||||
// fell back to a higher-precision spawn timestamp. Treat it as ambiguous and
|
||||
// fail closed instead of declaring the live PID recycled.
|
||||
return Math.abs(left.getTime() - right.getTime()) >= 1_000;
|
||||
// Shared with every other owner proof: a sub-second gap is ambiguous.
|
||||
return processStartsDefinitelyDiffer(left, right);
|
||||
}
|
||||
|
||||
export async function evaluateNativeControllerTakeover(input: {
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
nativeRunPreparationStarts,
|
||||
recordFailedSkillPreparation,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
resolveNativeTimeToFirstTokenStart,
|
||||
NATIVE_RUN_SPAN_EVENT_TYPE,
|
||||
NATIVE_RUN_TRACE_SCHEMA_VERSION,
|
||||
} from "./native-run-trace.js";
|
||||
@@ -504,3 +505,102 @@ it.each(["start", "context", "attribute", "status", "end"])("keeps execution and
|
||||
expect(failedWork).toHaveBeenCalledOnce();
|
||||
for (const span of spans) expect(span.end).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
describe("time to first token anchor", () => {
|
||||
const prep = (name: string, startedAtMs: number) => ({
|
||||
name, parentName: "task.provider_session", startedAtMs, endedAtMs: startedAtMs + 5,
|
||||
});
|
||||
|
||||
it("falls back to run creation when no earlier accepted input is known", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 1_000, preparationSpans: [prep("heartbeat.queue", 1_000)], wakeRequestedAtMs: null,
|
||||
})).toEqual({ startedAtMs: 1_000, anchor: "run_created" });
|
||||
});
|
||||
|
||||
it("names the comment or answered question that already anchors the trace", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 900, preparationSpans: [prep("comment.to_run_created", 900)], wakeRequestedAtMs: 950,
|
||||
})).toEqual({ startedAtMs: 900, anchor: "comment" });
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 800, preparationSpans: [prep("question_response.to_run_created", 800)], wakeRequestedAtMs: null,
|
||||
})).toEqual({ startedAtMs: 800, anchor: "question_response" });
|
||||
});
|
||||
|
||||
it("names run creation when a comment was coalesced into an already-queued run", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 1_000,
|
||||
preparationSpans: [prep("heartbeat.queue", 1_000), prep("comment.to_run_created", 1_400)],
|
||||
wakeRequestedAtMs: null,
|
||||
})).toEqual({ startedAtMs: 1_000, anchor: "run_created" });
|
||||
});
|
||||
|
||||
it("moves back to an earlier accepted wake request (deferred or coalesced wakes)", () => {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 5_000, preparationSpans: [prep("heartbeat.queue", 5_000)], wakeRequestedAtMs: 2_000,
|
||||
})).toEqual({ startedAtMs: 2_000, anchor: "wake_request" });
|
||||
});
|
||||
|
||||
it("ignores invalid wake request times", () => {
|
||||
for (const wakeRequestedAtMs of [Number.NaN, -1, 6_000]) {
|
||||
expect(resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: 5_000, preparationSpans: [], wakeRequestedAtMs,
|
||||
})).toEqual({ startedAtMs: 5_000, anchor: "run_created" });
|
||||
}
|
||||
});
|
||||
|
||||
it("persists the anchor on the run-log span", async () => {
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
const trace = createNativeRunTrace({
|
||||
runId: "ttft-run", startedAtMs: 5_000,
|
||||
traceContext: createRecordingTraceContext().traceContext,
|
||||
onEvent: async (event) => { events.push(event); },
|
||||
});
|
||||
const start = resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: trace.traceStartedAtMs, preparationSpans: [], wakeRequestedAtMs: 4_000,
|
||||
});
|
||||
await trace.record({
|
||||
name: "task.time_to_first_token", parentName: "task.provider_session",
|
||||
startedAtMs: start.startedAtMs, endedAtMs: 7_000, attributes: { eventKind: "agentMessage", anchor: start.anchor },
|
||||
});
|
||||
const span = events.find((event) => (event.payload as Record<string, unknown>)?.span === "task.time_to_first_token");
|
||||
expect(span?.payload).toMatchObject({ durationMs: 3_000, startOffsetMs: -1_000, anchor: "wake_request", eventKind: "agentMessage" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("deferred diagnostics", () => {
|
||||
it("never blocks the caller and settles before the trace finishes", async () => {
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
const trace = createNativeRunTrace({
|
||||
runId: "deferred-run", startedAtMs: 1_000,
|
||||
traceContext: createRecordingTraceContext().traceContext,
|
||||
onEvent: async (event) => { events.push(event); },
|
||||
});
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((resolve) => { release = resolve; });
|
||||
trace.defer(async () => {
|
||||
await gate;
|
||||
await trace.record({ name: "task.time_to_first_token", parentName: "task.provider_session", startedAtMs: 1_000, endedAtMs: 2_000 });
|
||||
});
|
||||
// Later work keeps flowing while the deferred read is pending.
|
||||
await trace.record({ name: "response.first_nonempty_text", startedAtMs: 2_000, endedAtMs: 2_000 });
|
||||
const span = (name: string) => events.some((event) => (event.payload as Record<string, unknown>)?.span === name);
|
||||
expect(span("response.first_nonempty_text")).toBe(true);
|
||||
expect(span("task.time_to_first_token")).toBe(false);
|
||||
const finished = trace.finish("ok");
|
||||
release();
|
||||
await finished;
|
||||
const names = events.map((event) => (event.payload as Record<string, unknown>)?.span);
|
||||
expect(names.indexOf("task.time_to_first_token")).toBeGreaterThan(-1);
|
||||
expect(names.indexOf("task.time_to_first_token")).toBeLessThan(names.indexOf("task.provider_session.measured"));
|
||||
});
|
||||
|
||||
it("ignores a failed deferred diagnostic and work deferred after finish", async () => {
|
||||
const trace = createNativeRunTrace({ runId: "deferred-fail", traceContext: createRecordingTraceContext().traceContext });
|
||||
trace.defer(async () => { throw new Error("diagnostic failed"); });
|
||||
await expect(trace.finish("ok")).resolves.toBeUndefined();
|
||||
const late = vi.fn(async () => {});
|
||||
trace.defer(late);
|
||||
await Promise.resolve();
|
||||
expect(late).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -106,6 +106,58 @@ export function buildNativeHeartbeatPreparationSpans(input: {
|
||||
];
|
||||
}
|
||||
|
||||
/** Which accepted input `task.time_to_first_token` is measured from. */
|
||||
export type NativeTimeToFirstTokenAnchor =
|
||||
| "question_response"
|
||||
| "comment"
|
||||
| "wake_request"
|
||||
| "run_created";
|
||||
|
||||
/**
|
||||
* Resolve the start of `task.time_to_first_token`: the earliest accepted input
|
||||
* that caused this run. A question answer or a wake comment is already a root
|
||||
* historical span. Otherwise the run row's creation time is the fallback. A
|
||||
* durable wake request accepted earlier (deferred, coalesced or retried wakes)
|
||||
* moves the start back to that request, so queueing before the run row exists
|
||||
* is part of the measured latency.
|
||||
*
|
||||
* Run-log only. The wake request time is read after the first deliverable
|
||||
* text, never before `turn.start`.
|
||||
*/
|
||||
export function resolveNativeTimeToFirstTokenStart(input: {
|
||||
traceStartedAtMs: number;
|
||||
preparationSpans: readonly NativeRunHistoricalSpan[];
|
||||
wakeRequestedAtMs: number | null;
|
||||
}): { startedAtMs: number; anchor: NativeTimeToFirstTokenAnchor } {
|
||||
const ingress = input.preparationSpans.find(
|
||||
(span) =>
|
||||
span.name === "question_response.to_run_created" ||
|
||||
span.name === "comment.to_run_created",
|
||||
);
|
||||
// The trace starts at the earliest preparation span. A comment coalesced
|
||||
// into an already-queued run starts after run creation, so it names the
|
||||
// start only when it is that earliest span.
|
||||
const ingressIsStart = ingress !== undefined && ingress.startedAtMs <= input.traceStartedAtMs;
|
||||
let anchor: NativeTimeToFirstTokenAnchor =
|
||||
!ingressIsStart
|
||||
? "run_created"
|
||||
: ingress.name === "question_response.to_run_created"
|
||||
? "question_response"
|
||||
: "comment";
|
||||
let startedAtMs = input.traceStartedAtMs;
|
||||
const wakeAt = input.wakeRequestedAtMs;
|
||||
if (
|
||||
wakeAt !== null &&
|
||||
Number.isFinite(wakeAt) &&
|
||||
wakeAt >= 0 &&
|
||||
wakeAt < startedAtMs
|
||||
) {
|
||||
startedAtMs = wakeAt;
|
||||
anchor = "wake_request";
|
||||
}
|
||||
return { startedAtMs, anchor };
|
||||
}
|
||||
|
||||
export function nativeRunPreparationStarts(
|
||||
spans: NativeRunHistoricalSpan[],
|
||||
nowMs: number,
|
||||
@@ -149,6 +201,7 @@ const NOOP_SPAN: SpanHandle = {
|
||||
|
||||
const SAFE_NATIVE_RUN_SPAN_ATTRIBUTE_KEYS = new Set([
|
||||
"adapter",
|
||||
"anchor",
|
||||
"bytesTransferred",
|
||||
"inputBytes", "toolSchemaBytes", "toolCount", "contextReused", "tokenDataAvailable",
|
||||
"connectionOwner",
|
||||
@@ -496,7 +549,28 @@ export function createNativeRunTrace(input: {
|
||||
}
|
||||
};
|
||||
|
||||
/**
|
||||
* Diagnostic work that must not block the caller (for example a read that
|
||||
* anchors a span). It runs detached; `finish` settles it before the root
|
||||
* span closes. The work must bound its own waits.
|
||||
*/
|
||||
const pendingDiagnostics = new Set<Promise<void>>();
|
||||
const defer = (work: () => Promise<void>): void => {
|
||||
if (ended) return;
|
||||
const pending: Promise<void> = Promise.resolve()
|
||||
.then(work)
|
||||
.catch(() => {
|
||||
// Diagnostics cannot change runner control flow.
|
||||
})
|
||||
.finally(() => {
|
||||
pendingDiagnostics.delete(pending);
|
||||
});
|
||||
pendingDiagnostics.add(pending);
|
||||
};
|
||||
|
||||
const finish = async (outcome: NativeRunSpanOutcome): Promise<void> => {
|
||||
if (ended) return;
|
||||
await Promise.all([...pendingDiagnostics]);
|
||||
if (ended) return;
|
||||
ended = true;
|
||||
const endedAtMs = Date.now();
|
||||
@@ -528,6 +602,7 @@ export function createNativeRunTrace(input: {
|
||||
run,
|
||||
record,
|
||||
measure,
|
||||
defer,
|
||||
finish,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,74 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { readNativeRunWakeRequestedAtMs } from "./native-run-wake-anchor.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = support.supported ? describe : describe.skip;
|
||||
|
||||
describeEmbeddedPostgres("native run wake anchor against PostgreSQL", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let locker!: ReturnType<typeof createDb>;
|
||||
const companyId = randomUUID();
|
||||
const agentId = randomUUID();
|
||||
const wakeupRequestId = randomUUID();
|
||||
const runId = randomUUID();
|
||||
const requestedAt = new Date("2026-11-07T10:00:00.000Z");
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-wake-anchor-");
|
||||
// One connection: a read left running after the timeout would starve the next query.
|
||||
db = createDb(database.connectionString, { maxConnections: 1 });
|
||||
locker = createDb(database.connectionString, { maxConnections: 2 });
|
||||
await db.insert(companies).values({ id: companyId, name: "Anchor", issuePrefix: `A${companyId.slice(0, 6)}` });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Anchor agent", role: "engineer", status: "idle", adapterType: "codex_local" });
|
||||
await db.insert(agentWakeupRequests).values({ id: wakeupRequestId, companyId, agentId, source: "assignment", status: "claimed", requestedAt });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, wakeupRequestId, status: "running" });
|
||||
}, 60_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await database?.cleanup();
|
||||
});
|
||||
|
||||
it("reads requested_at through the run's wake request", async () => {
|
||||
await expect(readNativeRunWakeRequestedAtMs(db, runId)).resolves.toBe(requestedAt.getTime());
|
||||
await expect(readNativeRunWakeRequestedAtMs(db, randomUUID())).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("cancels a blocked read on the server and returns its connection to the pool", async () => {
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>((resolve) => { release = resolve; });
|
||||
let locked!: () => void;
|
||||
const lockHeld = new Promise<void>((resolve) => { locked = resolve; });
|
||||
const holder = locker.transaction(async (tx) => {
|
||||
await tx.execute(sql`lock table agent_wakeup_requests in access exclusive mode`);
|
||||
locked();
|
||||
await gate;
|
||||
});
|
||||
try {
|
||||
await lockHeld;
|
||||
const startedAt = Date.now();
|
||||
await expect(readNativeRunWakeRequestedAtMs(db, runId, 300)).resolves.toBeNull();
|
||||
expect(Date.now() - startedAt).toBeLessThan(2_000);
|
||||
// The lock is still held. The single pooled connection must already be free,
|
||||
// and PostgreSQL must not still be waiting on the anchor read.
|
||||
const probeStartedAt = Date.now();
|
||||
const [probe] = await db.execute<{ ok: number }>(sql`select 1 as ok`);
|
||||
expect(Number(probe?.ok)).toBe(1);
|
||||
expect(Date.now() - probeStartedAt).toBeLessThan(1_000);
|
||||
const waiting = await locker.execute<{ count: string }>(sql`
|
||||
select count(*)::text as count from pg_stat_activity
|
||||
where state = 'active' and wait_event_type = 'Lock' and query ilike '%agent_wakeup_requests%'`);
|
||||
expect(Number(waiting[0]?.count ?? 0)).toBe(0);
|
||||
} finally {
|
||||
release();
|
||||
await holder;
|
||||
}
|
||||
}, 20_000);
|
||||
});
|
||||
@@ -0,0 +1,39 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
|
||||
import { readNativeRunWakeRequestedAtMs } from "./native-run-wake-anchor.js";
|
||||
|
||||
function fakeDb(result: () => Promise<unknown[]>, statements: string[] = []): Db {
|
||||
const chain = {
|
||||
select: () => chain, from: () => chain, innerJoin: () => chain, where: () => chain,
|
||||
limit: () => result(),
|
||||
execute: async (query: { queryChunks?: unknown[] }) => { statements.push(JSON.stringify(query.queryChunks ?? query)); return []; },
|
||||
};
|
||||
return { transaction: async (fn: (tx: typeof chain) => Promise<unknown>) => fn(chain) } as unknown as Db;
|
||||
}
|
||||
|
||||
describe("readNativeRunWakeRequestedAtMs", () => {
|
||||
it("returns the wake request acceptance time under a transaction-local statement timeout", async () => {
|
||||
const at = new Date("2026-11-07T10:00:00.000Z");
|
||||
const statements: string[] = [];
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(async () => [{ requestedAt: at }], statements), "run", 250))
|
||||
.resolves.toBe(at.getTime());
|
||||
expect(statements.join(" ")).toContain("statement_timeout");
|
||||
expect(statements.join(" ")).toContain("250ms");
|
||||
});
|
||||
|
||||
it("returns null when the run has no wake request", async () => {
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(async () => []), "run")).resolves.toBeNull();
|
||||
});
|
||||
|
||||
it("stops waiting at the timeout even before the database answers", async () => {
|
||||
const pending = new Promise<unknown[]>(() => {});
|
||||
const startedAt = Date.now();
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(() => pending), "run", 20)).resolves.toBeNull();
|
||||
expect(Date.now() - startedAt).toBeLessThan(1_000);
|
||||
});
|
||||
|
||||
it("never throws: a read failure only drops the diagnostic anchor", async () => {
|
||||
await expect(readNativeRunWakeRequestedAtMs(fakeDb(async () => { throw new Error("db down"); }), "run")).resolves.toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,56 @@
|
||||
import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
|
||||
/** A slow database never holds this diagnostic longer than this. */
|
||||
export const NATIVE_RUN_WAKE_ANCHOR_READ_TIMEOUT_MS = 2_000;
|
||||
|
||||
/**
|
||||
* When the durable wake request that created this run was accepted, in epoch
|
||||
* milliseconds. Diagnostic only: it anchors `task.time_to_first_token` to the
|
||||
* accepted input. Callers run it detached after the first deliverable text, so
|
||||
* it never adds a wait before `turn.start` or to event delivery.
|
||||
*
|
||||
* The read is bounded on both sides. PostgreSQL cancels the statement at
|
||||
* `timeoutMs` (a transaction-local `statement_timeout`, which also covers lock
|
||||
* waits), so a stalled read does not keep a shared connection busy after the
|
||||
* caller stops waiting. The caller also stops waiting at `timeoutMs`, which
|
||||
* covers a wait for a pool connection. Any failure returns null.
|
||||
*/
|
||||
export async function readNativeRunWakeRequestedAtMs(
|
||||
db: Db,
|
||||
runId: string,
|
||||
timeoutMs = NATIVE_RUN_WAKE_ANCHOR_READ_TIMEOUT_MS,
|
||||
): Promise<number | null> {
|
||||
const boundedMs = Math.max(1, Math.floor(timeoutMs));
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const timeout = new Promise<null>((resolve) => {
|
||||
timer = setTimeout(() => resolve(null), boundedMs);
|
||||
timer.unref?.();
|
||||
});
|
||||
try {
|
||||
return await Promise.race([readWakeRequestedAtMs(db, runId, boundedMs), timeout]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
async function readWakeRequestedAtMs(db: Db, runId: string, timeoutMs: number): Promise<number | null> {
|
||||
try {
|
||||
return await db.transaction(async (tx) => {
|
||||
await tx.execute(sql`select set_config('statement_timeout', ${`${timeoutMs}ms`}, true)`);
|
||||
const [row] = await tx
|
||||
.select({ requestedAt: agentWakeupRequests.requestedAt })
|
||||
.from(heartbeatRuns)
|
||||
.innerJoin(
|
||||
agentWakeupRequests,
|
||||
eq(agentWakeupRequests.id, heartbeatRuns.wakeupRequestId),
|
||||
)
|
||||
.where(eq(heartbeatRuns.id, runId))
|
||||
.limit(1);
|
||||
const at = row?.requestedAt?.getTime();
|
||||
return typeof at === "number" && Number.isFinite(at) ? at : null;
|
||||
});
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -139,6 +139,9 @@ vi.mock("../usage-receipts.js", async (importOriginal) => ({
|
||||
createRunUsageRecorder: usageRecorder.create,
|
||||
}));
|
||||
|
||||
const wakeAnchor = vi.hoisted(() => ({ read: vi.fn(async (_db: unknown, _runId: string): Promise<number | null> => null) }));
|
||||
vi.mock("./native-run-wake-anchor.js", () => ({ readNativeRunWakeRequestedAtMs: wakeAnchor.read }));
|
||||
|
||||
vi.mock("./native-usage-delivery.js", () => ({
|
||||
createNativeUsageDelivery: vi.fn(async () => ({ capture: async () => undefined })),
|
||||
}));
|
||||
@@ -8826,7 +8829,9 @@ describe("native resumed preparation timing", () => {
|
||||
const ttft = events.filter((event) => event.payload?.span === "task.time_to_first_token");
|
||||
expect(ttft).toHaveLength(1);
|
||||
expect(ttft[0]?.payload).toMatchObject({ parentSpan: "task.provider_session", startOffsetMs: 0,
|
||||
durationMs: ingress === "native_dispatch" ? 500 : 4_500, eventKind: "agentMessage", outcome: "ok" });
|
||||
durationMs: ingress === "native_dispatch" ? 500 : 4_500, eventKind: "agentMessage", outcome: "ok",
|
||||
anchor: ingress === "comment.to_run_created" ? "comment"
|
||||
: ingress === "question_response.to_run_created" ? "question_response" : "run_created" });
|
||||
expect(events.find((event) => event.payload?.span === "provider.time_to_first_agent_event")?.payload)
|
||||
.toMatchObject({ parentSpan: "agent.turn", eventKind: "agentMessage" });
|
||||
expect(JSON.stringify(ttft)).not.toContain("first text");
|
||||
@@ -8834,6 +8839,41 @@ describe("native resumed preparation timing", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it("anchors first-token timing at an earlier durable wake request without blocking later events", async () => {
|
||||
const startedAtMs = Date.now() - 5_000;
|
||||
const events: AdapterRuntimeEvent[] = [];
|
||||
const clock = vi.spyOn(Date, "now").mockReturnValue(startedAtMs + 4_000);
|
||||
let answerRead!: (value: number) => void;
|
||||
wakeAnchor.read.mockImplementationOnce(() => new Promise<number>((resolve) => { answerRead = resolve; }));
|
||||
const event = (eventType: string, payload: Record<string, unknown>) => ({
|
||||
eventType, payload, itemId: "message", emittedAt: new Date(startedAtMs + 2_000).toISOString(),
|
||||
}) as PrpEvent;
|
||||
state.execute.mockReset().mockImplementationOnce(async () => {
|
||||
for (const candidate of [event("turn.submitted", {}), event("turn.started", {})]) await accountingEvents.committed!(candidate);
|
||||
clock.mockReturnValue(startedAtMs + 4_500);
|
||||
await accountingEvents.committed!(event("item.delta", { kind: "agentMessage", text: "first text" }));
|
||||
// The anchor read is still pending; later events are not held behind it.
|
||||
await accountingEvents.committed!(event("item.delta", { kind: "agentMessage", text: "later text" }));
|
||||
expect(events.some((candidate) => candidate.payload?.span === "task.time_to_first_token")).toBe(false);
|
||||
answerRead(startedAtMs - 2_000);
|
||||
return { result: { summary: "cancelled" }, terminal: { runTerminalState: "cancelled" },
|
||||
turnId: "turn", normalizedSessionId: "session", providerSessionId: null,
|
||||
driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1 };
|
||||
});
|
||||
try {
|
||||
await executePaperclipNativeSession({
|
||||
db: leaseDb(), execution, runnerInstanceId: "runner",
|
||||
preparationSpans: [{ name: "heartbeat.queue", startedAtMs, endedAtMs: startedAtMs + 100 }],
|
||||
onEvent: async (event) => { events.push(event); },
|
||||
});
|
||||
} finally { clock.mockRestore(); }
|
||||
expect(wakeAnchor.read).toHaveBeenCalledWith(expect.anything(), execution.binding.runId);
|
||||
const ttft = events.filter((event) => event.payload?.span === "task.time_to_first_token");
|
||||
expect(ttft).toHaveLength(1);
|
||||
expect(ttft[0]?.payload).toMatchObject({ parentSpan: "task.provider_session", startOffsetMs: -2_000,
|
||||
durationMs: 6_500, anchor: "wake_request", eventKind: "agentMessage" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["item.delta", { kind: "reasoning", text: "thinking" }, "reasoning"],
|
||||
["item.completed", { kind: "reasoning", text: "thought" }, "reasoning"],
|
||||
|
||||
@@ -284,16 +284,19 @@ import {
|
||||
isNativeRunRootHistoricalSpan,
|
||||
nativeRunPreparationStarts,
|
||||
recordNativeProviderPerformanceSpans,
|
||||
resolveNativeTimeToFirstTokenStart,
|
||||
type NativeRunHistoricalSpan,
|
||||
type NativeRunSpanScope,
|
||||
type NativeRunTrace,
|
||||
} from "./native-run-trace.js";
|
||||
import { createNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js";
|
||||
import { readNativeRunWakeRequestedAtMs } from "./native-run-wake-anchor.js";
|
||||
import { noteNativeRunnerEvidence } from "./native-runner-evidence-signal.js";
|
||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||
import { registerLiveRunnerGoalController } from "../runner-goal-control-broker.js";
|
||||
import { applyRunnerGoalPrpEvent } from "../runner-goals.js";
|
||||
import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import { processExists, processStartsDefinitelyDiffer } from "../controller-process-identity.js";
|
||||
import {
|
||||
NativeRunnerOwnershipUnverifiedError,
|
||||
isNativeRunnerOwnershipHeld,
|
||||
@@ -579,21 +582,21 @@ const nativeRuntimeRequestResolutions = new Map<
|
||||
NativeRuntimeRequestResolution
|
||||
>();
|
||||
|
||||
async function verifiedRecoveryProcessIsAlive(input: {
|
||||
/**
|
||||
* True when the recorded runner still owns its PID. A start time within the
|
||||
* shared precision window is the same process: `ps -o lstart` on macOS has
|
||||
* whole-second precision, so an exact match would call a live runner gone.
|
||||
*/
|
||||
export async function verifiedRecoveryProcessIsAlive(input: {
|
||||
pid: number;
|
||||
startedAt: string;
|
||||
}): Promise<boolean> {
|
||||
try {
|
||||
process.kill(input.pid, 0);
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException | undefined)?.code !== "EPERM")
|
||||
return false;
|
||||
}
|
||||
if (!processExists(input.pid)) return false;
|
||||
try {
|
||||
const observed = await readProcessStartedAt(input.pid);
|
||||
return (
|
||||
observed !== null &&
|
||||
new Date(observed).getTime() === new Date(input.startedAt).getTime()
|
||||
!processStartsDefinitelyDiffer(input.startedAt, observed)
|
||||
);
|
||||
} catch {
|
||||
return false;
|
||||
@@ -10206,12 +10209,21 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
firstTokenRecorded = true;
|
||||
const textAt = Date.now();
|
||||
await trace.record({name:"response.first_nonempty_text",parentName:"native.session.execute",startedAtMs:textAt,endedAtMs:textAt});
|
||||
await trace.record({
|
||||
name: "task.time_to_first_token",
|
||||
parentName: "task.provider_session",
|
||||
startedAtMs: trace.traceStartedAtMs,
|
||||
endedAtMs: Date.now(),
|
||||
attributes: { eventKind: textKind },
|
||||
// Detached: the anchor read never delays this event, later events,
|
||||
// or turn.start (invariant 8). The trace settles it at finish.
|
||||
trace.defer(async () => {
|
||||
const ttftStart = resolveNativeTimeToFirstTokenStart({
|
||||
traceStartedAtMs: trace.traceStartedAtMs,
|
||||
preparationSpans,
|
||||
wakeRequestedAtMs: await readNativeRunWakeRequestedAtMs(input.db, input.execution.binding.runId),
|
||||
});
|
||||
await trace.record({
|
||||
name: "task.time_to_first_token",
|
||||
parentName: "task.provider_session",
|
||||
startedAtMs: ttftStart.startedAtMs,
|
||||
endedAtMs: textAt,
|
||||
attributes: { eventKind: textKind, anchor: ttftStart.anchor },
|
||||
});
|
||||
});
|
||||
}
|
||||
await observeAccountingEvent(event);
|
||||
|
||||
Reference in new issue
Block a user