mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
perf(runner): deliver transport events when they arrive instead of polling every 5 ms (#20862)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The server runs each native agent session through a runner transport. The transport reads the events that the runner commits to the session's control plane and delivers them to the driver. > - The transport found those events with a timer. Every live transport ran a pass every 5 ms, for as long as the session was open, whether or not an event had arrived. > - A CPU profile of the staging server shows the cost. The main thread was at 96% of one core, and this pass used 5.6% of it with about a dozen live sessions. The cost grows with the number of sessions and never stops. > - The same timer also delayed every event. An event waited up to 5 ms for the next pass before the driver saw it. Agent boot time and time to first token are measured requirements. > - This pull request makes the control plane tell the transport when a reader can see a change, and the transport delivers in the next task. The fixed timer is gone. > - The benefit is that an open session with nothing to deliver costs nothing on the main thread, and an event reaches the driver sooner than before. ## Linked Issues or Issue Description No public issue exists. The problem is described here. **What happened?** The server's main thread on a hosted instance stayed at 96% of one core, so every request and every agent turn waited behind it. A 30 second CPU profile of the live server (Node 24, about a dozen live native agent sessions) attributes 5.6% of the main thread to `#pumpEventsSafely` and 4.7% self time to `#pumpEvents` in `packages/paperclip-runner/src/live/runnerd-codex-transport.ts`. Three places started `setInterval(() => this.#pumpEventsSafely(), 5)`. Each pass scans the session's committed-event window, so the cost is `sessions x 200 passes per second x window size`, with nothing to deliver. **Expected behavior** An open session that has no event to deliver does no periodic work. An event that arrives is delivered at once, not at the next timer tick. **Steps to reproduce** 1. `cd packages/paperclip-runner` 2. `PAPERCLIP_TRANSPORT_PUMP_BENCHMARK=1 pnpm exec vitest run src/live/runnerd-transport-pump.benchmark.test.ts` 3. Read the `TRANSPORT_PUMP_BENCHMARK` lines. `stateReadsPerSessionPerSecond` counts how often an open, idle transport reads the control plane. `deliveryWaitMs` is the time from the moment a reader can see a committed event to the moment the transport delivers it. **Paperclip version or commit** Present on master at `c8ae02e5ce`. The profile is from `556b249897`. **Deployment mode** Hosted server with native agent sessions. The same code runs in local development. ## What Changed - `DurablePrpControlPlane` has a new `onReadableStateChange(listener)`. It fires when the state that a reader of `store` can see may have changed. The control plane calls it after every stored-state replacement or save, after a saved checkpoint releases events that were waiting for it, and after an authenticated runner moves the recovered cursor. - The transport subscribes when it creates its control plane. A change wakes one delivery pass in the next task (`setImmediate`). The pass never runs inside the control plane's own commit, and several changes in one task share one pass. This keeps the existing order: the control plane finishes its commit before a reader reacts. - The three `setInterval(..., 5)` timers are gone. `#startPump` and `#stopPump` replace them and keep the same lifecycle: the pump runs from session start, and it stops on release, controller retirement, failure, and close. - The delivery pass now reports what it left behind. Events that the transport holds back itself (a pending turn start, a pending recovery binding, a trace write that must be retried) are retried every 5 ms, as before, and only while they are held. This is the only timer that remains. - A pass that stops after it exposes `turn/started` (so the driver can bind the turn first) now continues in the next task. Before, the events after it waited for the next 5 ms tick. - `attachRun` wakes the pump when it moves the transport to the new run identity. - New regression test `src/live/runnerd-codex-transport-pump.test.ts`, new opt-in benchmark `src/live/runnerd-transport-pump.benchmark.test.ts`, a paced provider fixture, and a test-only probe that records when each committed event becomes readable. How this meets `rules/sleep-aware-background-work.mdc`: | Item | Answer | | --- | --- | | Trigger | The control plane's readable-state change. No timer. | | Source of truth | The control plane's committed-event window and the transport's delivery cursor. A notification is a hint to look at them. | | Pending-work predicate | The pass returns `held` or `next_task`, or a trace write is waiting. | | Next deadline or retry | 5 ms, only while the predicate is true. `next_task` uses one `setImmediate`. | | Quiet condition | Nothing is held. Then the transport has no timer and no scheduled task. Release, retirement, failure, and close stop it completely. | ## Verification **Reproduction and before/after measurement** Run from `packages/paperclip-runner`: ```sh PAPERCLIP_TRANSPORT_PUMP_BENCHMARK=1 pnpm exec vitest run src/live/runnerd-transport-pump.benchmark.test.ts ``` The benchmark opens 12 real transports against `runnerd` and a credential-free provider fixture. It measures the open sessions while nothing arrives. Then all 12 sessions stream one turn at the same time (40 events each, one every 15 ms). "Before" is the unchanged source at `e2cf9a670f`, which was the base of this branch when the measurements were taken. "After" is this change on that base. Each cell shows two runs, and the runs alternate between before and after. The host was a shared macOS arm64 machine with other work running (load average 8 to 10), so absolute numbers carry that noise. External checkpoint persistence. This is the mode that the server uses. | Measure | Before | After | | --- | --- | --- | | Idle: control-plane state reads per session per second | 364 / 365 | 0 / 0 | | Idle: event-loop utilization, 12 new sessions | 1.5% / 1.5% | 0.0% / 0.0% | | Idle: event-loop utilization, 49 retained events per session | 2.1% / 2.2% | 0.0% / 0.0% | | Idle: process CPU, 49 retained events per session | 12.8 / 11.8 ms per second | 0.05 / 0.04 ms per second | | Delivery wait, mean (480 events) | 3.48 / 4.22 ms | 1.91 / 1.81 ms | | Delivery wait, p50 | 3.16 / 3.97 ms | 0.83 / 0.58 ms | | Delivery wait, p95 | 7.78 / 8.49 ms | 6.36 / 7.47 ms | | Delivery wait, max | 8.84 / 9.23 ms | 9.07 / 8.86 ms | | Delivery wait of the turn's terminal event, p50 | 2.37 / 2.58 ms | 0.08 / 0.05 ms | The idle cost grows with the retained event window, as the profile predicts. With `PAPERCLIP_TRANSPORT_PUMP_DELTAS=400` (409 retained events per session, one run each): idle event-loop utilization 3.4% before and 0.0% after, idle process CPU 21.5 ms per second before and 0.08 after. Delivery wait in that run: mean 5.93 ms before and 3.55 ms after, p95 16.27 ms before and 8.03 ms after, max 66.94 ms before and 14.04 ms after. Delivery wait is the time from the moment a reader can see a committed event to the moment the transport hands it to the driver. It is equal or better at every point. The p95 and the max did not move much. They belong to the first events of each turn, which wait for the turn start response. That wait is a separate 10 ms poll (`waitForRunnerCommand`), and this pull request does not change it. File persistence (the local journal, which writes and syncs on the main thread): | Measure | Before | After | | --- | --- | --- | | Idle: state reads per session per second | 363 / 364 | 0 / 0 | | Delivery wait, 12 sessions, mean | 10.33 / 13.76 ms | 11.88 / 12.73 ms | | Delivery wait, 12 sessions, p50 | 6.04 / 6.87 ms | 6.34 / 6.20 ms | | Delivery wait, 4 sessions and 400 events each, mean | 4.75 ms | 2.39 ms | | Delivery wait, 4 sessions and 400 events each, p50 / p95 | 0.07 / 22.99 ms | 0.06 / 11.80 ms | With 12 sessions the file journal saturates the main thread with its own synchronous writes, and the wait is the time behind those writes. Three more before-runs of that case gave means of 9.0, 14.0 and 20.2 ms, so the two columns are the same within the noise. With 4 sessions the journal does not saturate and the after column is better. **Regression test (red, then green)** ```sh pnpm exec vitest run src/live/runnerd-codex-transport-pump.test.ts ``` - `stops looking for events while an open session has none to deliver` (file and external persistence). On the unchanged source: `expected 708 to be less than 100` and `expected 732 to be less than 100` state reads in a 2 second idle window. With the fix: pass. - `delivers each committed event in the next task, without a timer tick` (file and external persistence). The test freezes every interval timer, runs a full turn, and checks that each event is delivered before a task that the test queues at the moment the event becomes readable. This is an order check and does not depend on the speed of the host. On the unchanged source: `the turn's events were delivered without a periodic poll: expected false to be true`. With the fix: pass. - Result: 4 of 4 failed on the unchanged source, 4 of 4 passed three times with the fix, and 4 of 4 passed again after the rebase. **Other local checks** - `node ./node_modules/typescript/bin/tsc -p tsconfig.json --noEmit` and `-p tsconfig.test.json --noEmit` in `packages/paperclip-runner`: pass on the rebased commit. - `pnpm exec vitest run src/live/runnerd-codex-transport` (15 files, 455 tests). These suites cover attachment, recovery, warm transition, suspension, and release and resume on real `runnerd` processes. Before the rebase, on a quieter host: all passed. On the rebased commit, at a host load average of 15 to 20: 454 passed and 1 failed. The failure is the terminal-replay row that is described below. - `pnpm exec vitest run src/live src/control-plane src/native-session-runtime.test.ts src/backends src/drivers/dot/runnerd-dot-driver.test.ts src/drivers/muse/runnerd-muse-driver.test.ts src/drivers/codex/codex-protocol-integrity.test.ts --maxWorkers=3` (81 files): 1566 passed, 32 failed, 6 skipped. The failures: - 30 failures in `runnerd-acpx-permission-ordering.test.ts` and `runnerd-acpx-permission-forwarding.test.ts`. Both replace the control plane with a fixture that appended events in place and relied on the 5 ms poll to notice them. The fixture now implements `onReadableStateChange` and notifies its readers when it emits. Both files pass (30 of 30). - 1 failure in `runnerd-codex-transport.test.ts`, `stops a lost runnerd's provider and escaped tool job from its lineage journal (journal=true)`: `runner did not durably suspend before checkpoint`. It passed 3 of 3 when run again alone. - 1 failure in `runnerd-codex-transport-settlement.test.ts`, the terminal-replay row of `settles only retained control authority without starting another provider turn`: `native_cleanup_maintenance_unproven`, with events still in the runner outbox. It failed again in 2 runs of that test group and in the rebased run above, each time while the host load average was 11 to 20. At a load average near 8 it passed 3 of 3 alone and 1 of 1 as a group with the fix. The unchanged master source (`c8ae02e5ce`) gives the same result: it passed 2 of 2 alone and 1 of 1 as a group at a load average near 8, and the same row failed with the same error (`native_cleanup_maintenance_unproven`, 27 events still in the runner outbox) in a group run at a load average of 11 to 15. So this row is sensitive to host load with or without this change. The failing step is the retained-session maintenance path. That path builds its own control plane, has no transport pump, has no listener on the new notification, and has a fixed 30 second deadline. - `pnpm verify` returns exit 2 (broader verification required, because of the new fixture file). CI supplies the broad coverage. - Not run locally: the server test suites, the repository-wide typecheck, and the build. CI ran them on `4bce83f50a`: 59 checks passed and 4 were skipped (Storybook deploy, Storybook visual regression, Sandbox provider tests, Select Vitest files). ## Risks - **A change that does not notify.** With no periodic pass, an event is delivered only when something wakes the pump. Every wake comes from one of a small, closed set: the five places where the store replaces or saves its state, the checkpoint publication, the recovered-cursor update, and the transport's own identity and start points. A new code path that changes readable state without one of these would leave an event undelivered until the next change. The regression test runs a full turn with every interval timer frozen, in both persistence modes, so a missing wake on the normal path fails the test. The existing transport suites (attachment, recovery, warm transition, suspension, release and resume) now run with no timer that could hide a missing wake. - **Delivery in the next task, not later.** An event now reaches the driver up to 5 ms earlier. The pass still runs in its own task after the commit, and the same gates hold a terminal event until the turn start response is released. The order of events does not change. - **No protocol, schema, Rust runner, or provider pack change.** The contract fingerprint inputs are untouched. Saved state from an older release resumes the same way. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Fable 5.2 (`claude-fable-5-2`), Anthropic, in Claude Code. Tool use and local code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5.2 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Fable 5.2 authored and GitHub committed
1 parent
cda23f08ac
commit
76da2fec4f
8 files changed
+632
-21
No files matched your search
@@ -1653,6 +1653,8 @@ class DurableCoreStore {
|
||||
#writeRevision = 0;
|
||||
#externallyPersisted = false;
|
||||
revision = 0;
|
||||
/** Called after the stored state is replaced or saved. */
|
||||
onChange: (() => void) | null = null;
|
||||
#pendingEvent: {
|
||||
buildCandidate: (state: StoredCoreState) => StoredCoreState | null;
|
||||
isCurrentOwner: () => boolean;
|
||||
@@ -1714,6 +1716,7 @@ class DurableCoreStore {
|
||||
this.#state.checkpointEventPolicy = "control-v1";
|
||||
this.#externallyPersisted = true;
|
||||
this.revision += 1;
|
||||
this.onChange?.();
|
||||
}
|
||||
|
||||
save(): void {
|
||||
@@ -1732,6 +1735,7 @@ class DurableCoreStore {
|
||||
this.#state = candidate;
|
||||
if (pending && withEvent) pending.committed = true;
|
||||
this.revision += 1;
|
||||
this.onChange?.();
|
||||
} catch (error) {
|
||||
this.#writeIndeterminate = true;
|
||||
throw error;
|
||||
@@ -1757,6 +1761,7 @@ class DurableCoreStore {
|
||||
this.#state = candidate;
|
||||
++this.#writeRevision;
|
||||
if (checkpointRequired) this.revision += 1;
|
||||
this.onChange?.();
|
||||
return true;
|
||||
}
|
||||
await withStagedPrivateWrite(this.path, `${JSON.stringify(candidate)}\n`, (temporary) => {
|
||||
@@ -1768,6 +1773,7 @@ class DurableCoreStore {
|
||||
++this.#writeRevision;
|
||||
pending.committed = true;
|
||||
if (checkpointRequired) this.revision += 1;
|
||||
this.onChange?.();
|
||||
});
|
||||
// Explicit authority commits do not absorb pending events. On conflict,
|
||||
// leave the event unacknowledged for replay instead of rewriting in a loop.
|
||||
@@ -1801,6 +1807,7 @@ class DurableCoreStore {
|
||||
}
|
||||
this.#state = candidate;
|
||||
if (checkpointRequired) this.revision += 1;
|
||||
this.onChange?.();
|
||||
} catch (error) {
|
||||
// Rename may already have succeeded before directory fsync failed.
|
||||
// Never overwrite that possibly durable receipt using stale memory.
|
||||
@@ -2091,6 +2098,7 @@ export class DurablePrpControlPlane {
|
||||
#persistenceTail = Promise.resolve();
|
||||
#sendQueues = new Map<AuthorityConnection, Promise<void>>();
|
||||
#recoveredCursor: { runId: string; sourceSeq: number } | null = null;
|
||||
#readableStateListeners = new Set<() => void>();
|
||||
|
||||
constructor(options: DurablePrpControlPlaneOptions) {
|
||||
if (
|
||||
@@ -2111,6 +2119,7 @@ export class DurablePrpControlPlane {
|
||||
options.stateDirectory,
|
||||
options.identity,
|
||||
);
|
||||
this.#store.onChange = () => this.#readableStateChanged();
|
||||
if (this.#store.state.checkpointEventPolicy === "control-v1") {
|
||||
this.#recoveredCursor = { runId: options.identity.runId, sourceSeq: this.#store.state.ackedSourceSeq };
|
||||
}
|
||||
@@ -2143,6 +2152,31 @@ export class DurablePrpControlPlane {
|
||||
return this.#publishedStore ?? this.#store;
|
||||
}
|
||||
|
||||
/**
|
||||
* Tell a reader of `store` that what it can see may have changed: an event
|
||||
* or command was committed, or a saved checkpoint released the events that
|
||||
* were waiting for it. This is a hint, and it can arrive partway through the
|
||||
* change, so the reader must look at the state in a later task. Returns the
|
||||
* function that removes the listener.
|
||||
*/
|
||||
onReadableStateChange(listener: () => void): () => void {
|
||||
this.#readableStateListeners.add(listener);
|
||||
return () => {
|
||||
this.#readableStateListeners.delete(listener);
|
||||
};
|
||||
}
|
||||
|
||||
#readableStateChanged(): void {
|
||||
for (const listener of this.#readableStateListeners) {
|
||||
// A reader can never fail the commit that woke it.
|
||||
try {
|
||||
listener();
|
||||
} catch {
|
||||
// The reader owns its own failure handling.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#capturePublishedState(snapshot: DurablePrpCheckpoint): StoredCoreState {
|
||||
const header = JSON.parse(snapshot.records.get("header")!) as CheckpointHeader;
|
||||
const cursor = JSON.parse(snapshot.records.get("cursor")!) as { ackedSourceSeq: number; replayDeliveries: number };
|
||||
@@ -2227,6 +2261,7 @@ export class DurablePrpControlPlane {
|
||||
this.#publishedState = published;
|
||||
this.#persistedRevision = revision;
|
||||
this.#checkpointPersistenceFailed = false;
|
||||
this.#readableStateChanged();
|
||||
} catch (error) {
|
||||
this.#checkpointPersistenceFailed = true;
|
||||
throw error;
|
||||
@@ -3795,6 +3830,7 @@ export class DurablePrpControlPlane {
|
||||
const cursor = Math.max(this.#store.state.ackedSourceSeq, pending.acknowledgedSourceSeq);
|
||||
this.#store.state.ackedSourceSeq = cursor;
|
||||
this.#recoveredCursor = { runId: this.#identity.runId, sourceSeq: cursor };
|
||||
this.#readableStateChanged();
|
||||
}
|
||||
connection.pendingChallenge = null;
|
||||
connection.lease = lease;
|
||||
|
||||
@@ -23,6 +23,11 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
|
||||
readonly recoveredSourceSeq = 0;
|
||||
store: TestPeer["store"];
|
||||
connectUrl = "ws://fixture.invalid";
|
||||
readers = new Set<() => void>();
|
||||
onReadableStateChange(reader: () => void) {
|
||||
this.readers.add(reader);
|
||||
return () => { this.readers.delete(reader); };
|
||||
}
|
||||
constructor(options: { identity: DurableRecoveryIdentity }) {
|
||||
this.store = { state: { identity: options.identity, committedEvents: [], commands: [] } };
|
||||
peer.current = this;
|
||||
@@ -54,6 +59,7 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
|
||||
priority: 0, deliveryCount: 1, logicalEffectCount: 1,
|
||||
envelope: { turnId: this.store.state.identity.turnId, payload: { payload } },
|
||||
});
|
||||
for (const reader of this.readers) reader();
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
@@ -21,6 +21,11 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
|
||||
readonly recoveredSourceSeq = 0;
|
||||
store: TestPeer["store"];
|
||||
connectUrl = "ws://fixture.invalid";
|
||||
readers = new Set<() => void>();
|
||||
onReadableStateChange(reader: () => void) {
|
||||
this.readers.add(reader);
|
||||
return () => { this.readers.delete(reader); };
|
||||
}
|
||||
constructor(options: { identity: DurableRecoveryIdentity }) {
|
||||
this.store = { state: { identity: options.identity, committedEvents: [], commands: [] } };
|
||||
peer.current = this;
|
||||
@@ -51,6 +56,7 @@ vi.mock("../control-plane/durable-prp-control-plane.js", async (importOriginal)
|
||||
priority: 0, deliveryCount: 1, logicalEffectCount: 1,
|
||||
envelope: { turnId: this.store.state.identity.turnId, payload: { payload } },
|
||||
});
|
||||
for (const reader of this.readers) reader();
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { probeTransportPump, type ReadableEvent, type TransportPumpProbe } from "../../test/support/transport-pump-probe.js";
|
||||
import {
|
||||
createCapabilityRunnerdCodexTransport,
|
||||
defaultCapabilityRunnerdBinary,
|
||||
} from "./runnerd-codex-transport.js";
|
||||
|
||||
// A live transport used to look for committed events every 5 ms for as long as
|
||||
// it was open, so a dozen open sessions kept the server's main thread busy
|
||||
// with nothing to deliver, and every event waited for the next tick.
|
||||
const fixture = resolve(import.meta.dirname, "../../test/fixtures/fake-paced-codex-app-server.mjs");
|
||||
const DELTAS = 40;
|
||||
|
||||
async function openTransport(
|
||||
persistence: "file" | "external",
|
||||
onReadable?: (event: ReadableEvent) => void,
|
||||
) {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-transport-pump-"));
|
||||
let probe: TransportPumpProbe | null = null;
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(),
|
||||
codexCommand: process.execPath,
|
||||
codexArgs: [fixture, String(DELTAS), "15"],
|
||||
environment: {},
|
||||
stateDirectory: join(root, "session"),
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
|
||||
controlPlaneRegistration: async (core) => {
|
||||
probe = probeTransportPump(core, persistence, onReadable);
|
||||
return probe.registration();
|
||||
},
|
||||
});
|
||||
try {
|
||||
await bundle.transport.request("thread/start", { cwd: root, model: "fixture-no-model", dynamicTools: [] });
|
||||
} catch (error) {
|
||||
await bundle.transport.close().catch(() => undefined);
|
||||
await rm(root, { recursive: true, force: true });
|
||||
throw error;
|
||||
}
|
||||
return {
|
||||
transport: bundle.transport,
|
||||
probe: probe!,
|
||||
async close() {
|
||||
await bundle.transport.close().catch(() => undefined);
|
||||
await rm(root, { recursive: true, force: true });
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe.each(["file", "external"] as const)("runnerd transport event pump with %s persistence", (persistence) => {
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("stops looking for events while an open session has none to deliver", async () => {
|
||||
const session = await openTransport(persistence);
|
||||
try {
|
||||
const before = session.probe.stateReads();
|
||||
await new Promise((elapsed) => setTimeout(elapsed, 2_000));
|
||||
// The 5 ms poll took the state several hundred times a second. Without
|
||||
// it, only the tail of session startup reads the state in this window.
|
||||
expect(session.probe.stateReads() - before).toBeLessThan(100);
|
||||
|
||||
// The quiet session still delivers the first event that arrives.
|
||||
const notifications = session.transport.notifications()[Symbol.asyncIterator]();
|
||||
await session.transport.request("turn/start", { input: [{ type: "text", text: "Stream the paced fixture." }] });
|
||||
for (;;) {
|
||||
const next = await notifications.next();
|
||||
expect(next.done).not.toBe(true);
|
||||
if (next.value.method === "turn/completed") break;
|
||||
}
|
||||
} finally {
|
||||
await session.close();
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it("delivers each committed event in the next task, without a timer tick", async () => {
|
||||
// No interval ever fires in this test, so nothing here can be delivered by
|
||||
// a periodic poll.
|
||||
vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
|
||||
// The control plane wakes the transport before this probe sees the event,
|
||||
// so a task queued here runs after the transport's delivery task. That
|
||||
// order does not depend on the host's speed.
|
||||
const overtaken = new Set<number>();
|
||||
const pending = new Map<number, NodeJS.Immediate>();
|
||||
const session = await openTransport(persistence, (event) => {
|
||||
if (event.eventType !== "item.delta") return;
|
||||
pending.set(event.sourceSeq, setImmediate(() => {
|
||||
pending.delete(event.sourceSeq);
|
||||
overtaken.add(event.sourceSeq);
|
||||
}));
|
||||
});
|
||||
try {
|
||||
const notifications = session.transport.notifications()[Symbol.asyncIterator]();
|
||||
await session.transport.request("turn/start", { input: [{ type: "text", text: "Stream the paced fixture." }] });
|
||||
// Events committed before this point can wait for the start response.
|
||||
const startAnswered = performance.now();
|
||||
const deliveredInOrder: boolean[] = [];
|
||||
let completed = false;
|
||||
let stalled: ReturnType<typeof setTimeout> | undefined;
|
||||
await Promise.race([
|
||||
(async () => {
|
||||
for (;;) {
|
||||
const next = await notifications.next();
|
||||
if (next.done) return;
|
||||
if (next.value.method === "item/agentMessage/delta") {
|
||||
const event = session.probe.readable.filter((candidate) => candidate.eventType === "item.delta")[deliveredInOrder.length]!;
|
||||
clearImmediate(pending.get(event.sourceSeq));
|
||||
pending.delete(event.sourceSeq);
|
||||
deliveredInOrder.push(event.at < startAnswered || !overtaken.has(event.sourceSeq));
|
||||
}
|
||||
if (next.value.method === "turn/completed") {
|
||||
completed = true;
|
||||
return;
|
||||
}
|
||||
}
|
||||
})(),
|
||||
new Promise<void>((elapsed) => { stalled = setTimeout(elapsed, 10_000); }),
|
||||
]);
|
||||
clearTimeout(stalled);
|
||||
|
||||
expect(completed, "the turn's events were delivered without a periodic poll").toBe(true);
|
||||
expect(deliveredInOrder).toHaveLength(DELTAS);
|
||||
expect(deliveredInOrder.every(Boolean), "every event was delivered in the task after it became readable").toBe(true);
|
||||
// The assertion above covers the events streamed after the start response.
|
||||
expect(session.probe.readable.some((event) => event.eventType === "item.delta" && event.at >= startAnswered)).toBe(true);
|
||||
} finally {
|
||||
for (const task of pending.values()) clearImmediate(task);
|
||||
vi.useRealTimers();
|
||||
await session.close();
|
||||
}
|
||||
}, 30_000);
|
||||
});
|
||||
@@ -115,6 +115,8 @@ const packageRoot = resolve(fileURLToPath(new URL("../..", import.meta.url)));
|
||||
const executableSuffix = process.platform === "win32" ? ".exe" : "";
|
||||
const MAX_NOTIFICATION_COUNT = 2_048;
|
||||
const MAX_NOTIFICATION_BYTES = 4 * 1024 * 1024;
|
||||
// How often the event pump retries events that it is holding back itself.
|
||||
const PUMP_HELD_RETRY_MS = 5;
|
||||
const RUNNER_CLIENT_VERSION = "0.3.0";
|
||||
const RUNNER_BOOTSTRAP_TICKET_TTL_MS = 60_000;
|
||||
const RUNNER_RESTORE_STABLE_MS = 60_000;
|
||||
@@ -3950,7 +3952,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
#lostRunnerHandle: RunnerProcessHandle | null = null;
|
||||
#adoptedRunnerMonitor: NodeJS.Timeout | null = null;
|
||||
#adoptedRunnerAuthenticated = false;
|
||||
#pump: NodeJS.Timeout | null = null;
|
||||
/** Events are delivered only between session start and release, retirement, failure or close. */
|
||||
#pumpActive = false;
|
||||
#pumpWake: NodeJS.Immediate | null = null;
|
||||
#pumpRetry: NodeJS.Timeout | null = null;
|
||||
#stopFollowingCore: (() => void) | null = null;
|
||||
#eventSourceSeq = 0;
|
||||
#eventIdentity: DurableRecoveryIdentity | null = null;
|
||||
#sourceRecoveryIdentity: DurableRecoveryIdentity | null = null;
|
||||
@@ -4640,8 +4646,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
|
||||
#quietReleasedRunner(): void {
|
||||
if (this.#pump !== null) clearInterval(this.#pump);
|
||||
this.#pump = null;
|
||||
this.#stopPump();
|
||||
if (this.#adoptedRunnerMonitor !== null) clearInterval(this.#adoptedRunnerMonitor);
|
||||
this.#adoptedRunnerMonitor = null;
|
||||
}
|
||||
@@ -4768,10 +4773,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
await this.#awaitRegistrationReady(registration?.ready);
|
||||
this.#assertStartupOpen();
|
||||
}
|
||||
if (this.#pump === null) {
|
||||
this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
|
||||
this.#pump.unref?.();
|
||||
}
|
||||
if (!this.#pumpActive) this.#startPump();
|
||||
if (this.options.adoptExistingRunner && this.#adoptedRunnerMonitor === null) this.#watchAdoptedRunner(this.options.adoptExistingRunner);
|
||||
}
|
||||
|
||||
@@ -5017,6 +5019,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#eventSourceSeq = 0;
|
||||
this.#deferredTurnStartEvents = [];
|
||||
this.#durableTurnId = desired.turnId;
|
||||
this.#wakePump();
|
||||
if (!previousReleased) await previousRelease?.();
|
||||
previousReleased = true;
|
||||
this.#controlPlaneRelease = registration?.release ?? null;
|
||||
@@ -5101,6 +5104,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#pendingDriverTraceInterpretations.length < 4_096
|
||||
) {
|
||||
this.#pendingDriverTraceInterpretations.push(structuredClone(input));
|
||||
this.#retryPumpWhile(true);
|
||||
} else if (traceResult === "retry") {
|
||||
this.#traceRehydrationSpoolOverflow = true;
|
||||
}
|
||||
@@ -5378,8 +5382,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
|
||||
async #retireControllerForRecovery(deadline: number): Promise<void> {
|
||||
if (this.#pump !== null) clearInterval(this.#pump);
|
||||
this.#pump = null;
|
||||
this.#stopPump();
|
||||
if (this.#adoptedRunnerMonitor !== null)
|
||||
clearInterval(this.#adoptedRunnerMonitor);
|
||||
this.#adoptedRunnerMonitor = null;
|
||||
@@ -5614,8 +5617,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#pendingDriverTraceInterpretations = [];
|
||||
this.#traceFrameIndex.clear();
|
||||
}
|
||||
if (this.#pump !== null) clearInterval(this.#pump);
|
||||
this.#pump = null;
|
||||
this.#stopPump();
|
||||
this.#stopFollowingCore?.();
|
||||
this.#stopFollowingCore = null;
|
||||
if (this.#adoptedRunnerMonitor !== null)
|
||||
clearInterval(this.#adoptedRunnerMonitor);
|
||||
this.#adoptedRunnerMonitor = null;
|
||||
@@ -6089,6 +6093,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
connectionLeaseTtlMs: 60 * 60 * 1_000,
|
||||
});
|
||||
this.#core = core;
|
||||
this.#followCommittedEvents(core);
|
||||
this.#carriedCommandCount = core.store.state.commands.length;
|
||||
// Externally launched runners own their state directory (for example in a
|
||||
// Daytona sandbox). Do not create an empty controller-side placeholder:
|
||||
@@ -6180,7 +6185,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#evidence.runnerPid = handle.child.pid ?? null;
|
||||
this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null;
|
||||
this.#publish();
|
||||
this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
|
||||
this.#startPump();
|
||||
await this.#waitCommand("run.prepare", undefined, undefined, true);
|
||||
this.#startupSteps?.advance("runner.process.launch", "provider.session.load");
|
||||
await this.#waitCommand("session.open", undefined, admissionDeadline, true);
|
||||
@@ -6665,6 +6670,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
connectionLeaseTtlMs: 60 * 60 * 1_000,
|
||||
});
|
||||
this.#core = core;
|
||||
this.#followCommittedEvents(core);
|
||||
this.#carriedCommandCount = core.store.state.commands.length;
|
||||
// Reusing resources still requires current environment authorization. The
|
||||
// optional hook avoids re-preparation after controller storage is lost.
|
||||
@@ -7133,7 +7139,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
}
|
||||
this.#assertStartupOpen();
|
||||
this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
|
||||
this.#startPump();
|
||||
if (adoptedRunner) {
|
||||
await this.#awaitAdoptedRunnerConnection(
|
||||
adoptedRunner,
|
||||
@@ -7807,7 +7813,28 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Deliver every committed event that is ready. The control plane wakes this
|
||||
* when a reader can see a change, so an event never waits for a timer and an
|
||||
* idle session runs none. Only what this transport holds back itself, behind
|
||||
* a pending turn start, a recovery binding or a trace write, is retried on a
|
||||
* timer, and only for as long as it is held.
|
||||
*/
|
||||
#pumpEvents(): void {
|
||||
const remaining = this.#deliverCommittedEvents();
|
||||
if (remaining === "next_task") this.#wakePump();
|
||||
this.#retryPumpWhile(
|
||||
remaining === "held" ||
|
||||
this.#pendingTraceRehydrations.length > 0 ||
|
||||
this.#pendingDriverTraceInterpretations.length > 0,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns what a later pass still has to deliver: nothing, events held
|
||||
* behind a gate, or events that only had to wait for the next task.
|
||||
*/
|
||||
#deliverCommittedEvents(): "none" | "held" | "next_task" {
|
||||
const core = this.#core;
|
||||
// The controller may activate its new epoch before attachRun observes the
|
||||
// confirmed handoff. Never consume either epoch with the other's cursor.
|
||||
@@ -7816,7 +7843,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
!this.#eventIdentity ||
|
||||
!recoveryIdentityMatches(core.store.state.identity, this.#eventIdentity)
|
||||
)
|
||||
return;
|
||||
return "none";
|
||||
this.#flushPendingTraceRehydrations();
|
||||
const events = core.store.state.committedEvents;
|
||||
for (;;) {
|
||||
@@ -7830,7 +7857,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
events.find((candidate) => candidate.sourceSeq > this.#eventSourceSeq);
|
||||
if (event === undefined) {
|
||||
this.#eventSourceSeq = Math.max(this.#eventSourceSeq, core.recoveredSourceSeq);
|
||||
return;
|
||||
return this.#deferredTurnStartEvents.length > 0 ? "held" : "none";
|
||||
}
|
||||
const fromDeferredQueue = deferredEvent !== undefined;
|
||||
if (!fromDeferredQueue && event.sourceSeq > this.#eventSourceSeq + 1) {
|
||||
@@ -7906,7 +7933,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#eventSourceSeq = event.sourceSeq;
|
||||
this.#deferredTurnStartEvents.push(structuredClone(event));
|
||||
}
|
||||
return;
|
||||
return "held";
|
||||
}
|
||||
// The control plane retains a sliding committed-event window. Track its
|
||||
// durable protocol cursor rather than an array index: once that array is
|
||||
@@ -8207,15 +8234,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
// The strict Codex driver must bind the provider turn from the response
|
||||
// before it observes terminal notifications. A fast provider can commit
|
||||
// start and terminal events in one durable batch, so stop after exposing
|
||||
// turn/started while the request is pending. The regular pump drains the
|
||||
// remaining events after the promise resolves.
|
||||
// turn/started while the request is pending. The next task drains the
|
||||
// remaining events; a terminal among them stays held until the response
|
||||
// is released.
|
||||
if (
|
||||
this.#turnStartResponsePending &&
|
||||
notifications.some(
|
||||
(notification) => notification.method === "turn/started",
|
||||
)
|
||||
) {
|
||||
return;
|
||||
return "next_task";
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -8394,6 +8422,47 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
}
|
||||
|
||||
#followCommittedEvents(core: DurablePrpControlPlane): void {
|
||||
this.#stopFollowingCore?.();
|
||||
this.#stopFollowingCore = core.onReadableStateChange(() => {
|
||||
if (this.#core === core) this.#wakePump();
|
||||
});
|
||||
}
|
||||
|
||||
#startPump(): void {
|
||||
this.#pumpActive = true;
|
||||
this.#wakePump();
|
||||
}
|
||||
|
||||
#stopPump(): void {
|
||||
this.#pumpActive = false;
|
||||
if (this.#pumpWake !== null) clearImmediate(this.#pumpWake);
|
||||
this.#pumpWake = null;
|
||||
this.#retryPumpWhile(false);
|
||||
}
|
||||
|
||||
/**
|
||||
* Deliver in the next task. The control plane can report a change partway
|
||||
* through its own commit, so the pump never runs inside that call; several
|
||||
* changes in one task share a single pass.
|
||||
*/
|
||||
#wakePump(): void {
|
||||
if (!this.#pumpActive || this.#pumpWake !== null) return;
|
||||
this.#pumpWake = setImmediate(() => {
|
||||
this.#pumpWake = null;
|
||||
if (this.#pumpActive) this.#pumpEventsSafely();
|
||||
});
|
||||
}
|
||||
|
||||
#retryPumpWhile(held: boolean): void {
|
||||
if (held && this.#pumpActive) {
|
||||
this.#pumpRetry ??= setInterval(() => this.#pumpEventsSafely(), PUMP_HELD_RETRY_MS);
|
||||
} else if (this.#pumpRetry !== null) {
|
||||
clearInterval(this.#pumpRetry);
|
||||
this.#pumpRetry = null;
|
||||
}
|
||||
}
|
||||
|
||||
/** True only when a runnerd this transport spawned on this host has exited
|
||||
* and its journalled provider and tool processes are proven stopped. */
|
||||
async #reapLostLocalRunnerLineage(closeDeadline: number): Promise<boolean> {
|
||||
@@ -8778,8 +8847,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
if (this.#failure !== null || this.#closed) return;
|
||||
this.#failure = error;
|
||||
this.#rejectFailureSignal(error);
|
||||
if (this.#pump !== null) clearInterval(this.#pump);
|
||||
this.#pump = null;
|
||||
this.#stopPump();
|
||||
try {
|
||||
this.#diagnostic(error.message);
|
||||
} finally {
|
||||
|
||||
@@ -0,0 +1,147 @@
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { expect, it } from "vitest";
|
||||
import { probeTransportPump, type TransportPumpProbe } from "../../test/support/transport-pump-probe.js";
|
||||
import {
|
||||
createCapabilityRunnerdCodexTransport,
|
||||
defaultCapabilityRunnerdBinary,
|
||||
} from "./runnerd-codex-transport.js";
|
||||
|
||||
// Opt in explicitly. Run from packages/paperclip-runner:
|
||||
// PAPERCLIP_TRANSPORT_PUMP_BENCHMARK=1 pnpm exec vitest run src/live/runnerd-transport-pump.benchmark.test.ts
|
||||
// PAPERCLIP_TRANSPORT_PUMP_SESSIONS (default 12) sets how many transports stay open.
|
||||
// PAPERCLIP_TRANSPORT_PUMP_DELTAS (default 40) sets how many events each turn
|
||||
// streams, and so how many committed events each session retains afterwards.
|
||||
//
|
||||
// It opens that many real transports against runnerd and a credential-free
|
||||
// provider fixture, then reports two things for each persistence mode:
|
||||
// - what the open transports cost the main thread while nothing arrives, and
|
||||
// - how long a committed event waits before the transport delivers it.
|
||||
// This is a local measurement that shares the host's background load, not a
|
||||
// CPU-isolated assertion; only delivery itself is asserted.
|
||||
const enabled = process.env.PAPERCLIP_TRANSPORT_PUMP_BENCHMARK === "1";
|
||||
const sessions = Number(process.env.PAPERCLIP_TRANSPORT_PUMP_SESSIONS ?? "12");
|
||||
const deltas = Number(process.env.PAPERCLIP_TRANSPORT_PUMP_DELTAS ?? "40");
|
||||
if (enabled && (!Number.isInteger(sessions) || sessions < 1 || sessions > 32)) {
|
||||
throw new Error("transport_pump_sessions_must_be_between_1_and_32");
|
||||
}
|
||||
if (enabled && (!Number.isInteger(deltas) || deltas < 1 || deltas > 512)) {
|
||||
throw new Error("transport_pump_deltas_must_be_between_1_and_512");
|
||||
}
|
||||
const fixture = resolve(import.meta.dirname, "../../test/fixtures/fake-paced-codex-app-server.mjs");
|
||||
const DELTA_INTERVAL_MS = 15;
|
||||
const IDLE_SETTLE_MS = 1_500;
|
||||
const IDLE_WINDOW_MS = 3_000;
|
||||
|
||||
const percentile = (sorted: number[], fraction: number) =>
|
||||
sorted[Math.min(sorted.length - 1, Math.floor(sorted.length * fraction))]!;
|
||||
const round = (value: number) => Math.round(value * 1_000) / 1_000;
|
||||
|
||||
it.skipIf(!enabled).each(["file", "external"] as const)(
|
||||
"measures idle pump cost and delivery wait with %s persistence",
|
||||
async (persistence) => {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-transport-pump-benchmark-"));
|
||||
const open: Array<{
|
||||
bundle: ReturnType<typeof createCapabilityRunnerdCodexTransport>;
|
||||
probe: () => TransportPumpProbe;
|
||||
}> = [];
|
||||
const measureIdle = async () => {
|
||||
await new Promise((settled) => setTimeout(settled, IDLE_SETTLE_MS));
|
||||
const reads = () => open.reduce((total, session) => total + session.probe().stateReads(), 0);
|
||||
const readsBefore = reads();
|
||||
const cpuBefore = process.cpuUsage();
|
||||
const loopBefore = performance.eventLoopUtilization();
|
||||
const started = performance.now();
|
||||
await new Promise((elapsed) => setTimeout(elapsed, IDLE_WINDOW_MS));
|
||||
const seconds = (performance.now() - started) / 1_000;
|
||||
const loop = performance.eventLoopUtilization(loopBefore);
|
||||
const cpu = process.cpuUsage(cpuBefore);
|
||||
return {
|
||||
seconds: round(seconds),
|
||||
stateReadsPerSessionPerSecond: round((reads() - readsBefore) / seconds / sessions),
|
||||
eventLoopUtilization: round(loop.utilization),
|
||||
cpuMsPerSecond: round((cpu.user + cpu.system) / 1_000 / seconds),
|
||||
};
|
||||
};
|
||||
try {
|
||||
for (let index = 0; index < sessions; index += 1) {
|
||||
let probe: TransportPumpProbe | null = null;
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(),
|
||||
codexCommand: process.execPath,
|
||||
codexArgs: [fixture, String(deltas), String(DELTA_INTERVAL_MS)],
|
||||
environment: {},
|
||||
stateDirectory: join(root, `session-${index}`),
|
||||
lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null },
|
||||
controlPlaneRegistration: async (core) => {
|
||||
probe = probeTransportPump(core, persistence);
|
||||
return probe.registration();
|
||||
},
|
||||
});
|
||||
open.push({ bundle, probe: () => probe! });
|
||||
await bundle.transport.request("thread/start", { cwd: root, model: "fixture-no-model", dynamicTools: [] });
|
||||
}
|
||||
|
||||
// Idle: every session is open and connected, and no event arrives.
|
||||
const idleFresh = await measureIdle();
|
||||
|
||||
// Streaming: every session runs one paced turn at the same time.
|
||||
const waits: number[] = [];
|
||||
const terminalWaits: number[] = [];
|
||||
await Promise.all(open.map(async ({ bundle, probe }) => {
|
||||
const delivered: number[] = [];
|
||||
let terminalAt: number | null = null;
|
||||
const notifications = bundle.transport.notifications()[Symbol.asyncIterator]();
|
||||
await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Stream the paced fixture." }] });
|
||||
for (;;) {
|
||||
const next = await notifications.next();
|
||||
if (next.done) throw new Error("transport_pump_stream_ended_without_terminal");
|
||||
if (next.value.method === "item/agentMessage/delta") delivered.push(performance.now());
|
||||
if (next.value.method === "turn/completed") {
|
||||
terminalAt = performance.now();
|
||||
break;
|
||||
}
|
||||
}
|
||||
const readable = probe().readable;
|
||||
const streamed = readable.filter((event) => event.eventType === "item.delta");
|
||||
const terminal = readable.find((event) => event.eventType === "turn.completed");
|
||||
expect(delivered).toHaveLength(deltas);
|
||||
expect(streamed).toHaveLength(deltas);
|
||||
expect(terminal).toBeDefined();
|
||||
streamed.forEach((event, index) => waits.push(delivered[index]! - event.at));
|
||||
terminalWaits.push(terminalAt! - terminal!.at);
|
||||
}));
|
||||
waits.sort((left, right) => left - right);
|
||||
terminalWaits.sort((left, right) => left - right);
|
||||
|
||||
// Idle again: each session now retains the turn's committed events.
|
||||
const idleAfterTurn = await measureIdle();
|
||||
|
||||
process.stdout.write(`TRANSPORT_PUMP_BENCHMARK ${JSON.stringify({
|
||||
schema: "paperclip.transport_pump_benchmark.v1",
|
||||
persistence,
|
||||
sessions,
|
||||
retainedEventsPerSession: open[0]!.probe().readable.length,
|
||||
idleFresh,
|
||||
idleAfterTurn,
|
||||
deliveryWaitMs: {
|
||||
samples: waits.length,
|
||||
mean: round(waits.reduce((total, wait) => total + wait, 0) / waits.length),
|
||||
p50: round(percentile(waits, 0.5)),
|
||||
p95: round(percentile(waits, 0.95)),
|
||||
max: round(waits.at(-1)!),
|
||||
},
|
||||
terminalDeliveryWaitMs: {
|
||||
samples: terminalWaits.length,
|
||||
p50: round(percentile(terminalWaits, 0.5)),
|
||||
max: round(terminalWaits.at(-1)!),
|
||||
},
|
||||
})}\n`);
|
||||
} finally {
|
||||
await Promise.all(open.map(({ bundle }) => bundle.transport.close().catch(() => undefined)));
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
120_000,
|
||||
);
|
||||
@@ -0,0 +1,110 @@
|
||||
// Credential-free, deterministic provider that streams one delta at a time.
|
||||
// No network, model, tool execution, or user files are used by this fixture.
|
||||
// Usage: node fake-paced-codex-app-server.mjs <deltaCount> <intervalMs>
|
||||
import { createInterface } from "node:readline";
|
||||
|
||||
const [countArg, intervalArg] = process.argv.slice(2);
|
||||
const deltaCount = Number(countArg);
|
||||
const intervalMs = Number(intervalArg);
|
||||
if (
|
||||
!Number.isInteger(deltaCount) || deltaCount < 1 || deltaCount > 512 ||
|
||||
!Number.isInteger(intervalMs) || intervalMs < 1 || intervalMs > 1_000
|
||||
) {
|
||||
throw new Error("paced_fixture_invalid_arguments");
|
||||
}
|
||||
const threadId = "paced-thread";
|
||||
const turns = new Map();
|
||||
let nextTurn = 0;
|
||||
const send = (value) => process.stdout.write(`${JSON.stringify(value)}\n`);
|
||||
|
||||
function stream(turnId) {
|
||||
let index = 0;
|
||||
const timer = setInterval(() => {
|
||||
if (turns.get(turnId) !== "inProgress") {
|
||||
clearInterval(timer);
|
||||
return;
|
||||
}
|
||||
if (index < deltaCount) {
|
||||
send({
|
||||
method: "item/agentMessage/delta",
|
||||
params: {
|
||||
threadId,
|
||||
turnId,
|
||||
itemId: `message-${turnId}`,
|
||||
delta: `${index.toString().padStart(4, "0")} `,
|
||||
},
|
||||
});
|
||||
index += 1;
|
||||
return;
|
||||
}
|
||||
clearInterval(timer);
|
||||
send({
|
||||
method: "item/completed",
|
||||
params: {
|
||||
threadId,
|
||||
turnId,
|
||||
item: { id: `message-${turnId}`, type: "agentMessage", text: "Fixture complete." },
|
||||
},
|
||||
});
|
||||
turns.set(turnId, "completed");
|
||||
send({
|
||||
method: "turn/completed",
|
||||
params: { threadId, turn: { id: turnId, status: "completed" } },
|
||||
});
|
||||
}, intervalMs);
|
||||
}
|
||||
|
||||
createInterface({ input: process.stdin }).on("line", (line) => {
|
||||
const { id, method, params = {} } = JSON.parse(line);
|
||||
if (!method || id === undefined) return;
|
||||
if (method === "initialize") {
|
||||
send({ id, result: { user: { sessionId: "paced-fixture" } } });
|
||||
} else if (["thread/start", "thread/resume", "thread/read"].includes(method)) {
|
||||
send({
|
||||
id,
|
||||
result: {
|
||||
model: "fixture-no-model",
|
||||
modelProvider: "fixture-no-provider",
|
||||
thread: {
|
||||
id: threadId,
|
||||
sessionId: "paced-fixture",
|
||||
status: { type: [...turns.values()].includes("inProgress") ? "active" : "idle" },
|
||||
...(params.includeTurns
|
||||
? { turns: [...turns].map(([turnId, status]) => ({ id: turnId, status })) }
|
||||
: {}),
|
||||
},
|
||||
},
|
||||
});
|
||||
} else if (method === "thread/turns/list") {
|
||||
const listed = [...turns].map(([turnId, status]) => ({
|
||||
id: turnId, status, items: [], itemsView: "notLoaded",
|
||||
}));
|
||||
if (params.sortDirection === "desc") listed.reverse();
|
||||
const offset = Number(params.cursor ?? 0);
|
||||
const limit = params.limit ?? 100;
|
||||
send({ id, result: {
|
||||
data: listed.slice(offset, offset + limit),
|
||||
nextCursor: offset + limit < listed.length ? String(offset + limit) : null,
|
||||
} });
|
||||
} else if (method === "turn/start") {
|
||||
const turnId = `paced-turn-${++nextTurn}`;
|
||||
turns.set(turnId, "inProgress");
|
||||
send({ id, result: { turn: { id: turnId, status: "inProgress" } } });
|
||||
send({
|
||||
method: "turn/started",
|
||||
params: { threadId, turn: { id: turnId, status: "inProgress" } },
|
||||
});
|
||||
stream(turnId);
|
||||
} else if (method === "turn/interrupt") {
|
||||
send({ id, result: {} });
|
||||
if (turns.get(params.turnId) === "inProgress") {
|
||||
turns.set(params.turnId, "interrupted");
|
||||
send({
|
||||
method: "turn/completed",
|
||||
params: { threadId, turn: { id: params.turnId, status: "interrupted" } },
|
||||
});
|
||||
}
|
||||
} else {
|
||||
send({ id, error: { code: -32601, message: "paced_fixture_unsupported_method" } });
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,103 @@
|
||||
import type { DurablePrpControlPlane } from "../../src/control-plane/durable-prp-control-plane.js";
|
||||
|
||||
export interface ReadableEvent {
|
||||
readonly sourceSeq: number;
|
||||
readonly eventType: string;
|
||||
/** `performance.now()` when a reader could first see this committed event. */
|
||||
readonly at: number;
|
||||
}
|
||||
|
||||
export interface TransportPumpProbe {
|
||||
/** How often anything took the control plane's readable state. A transport takes it on every pump pass. */
|
||||
stateReads(): number;
|
||||
/** Committed events in the order readers could first see them. */
|
||||
readonly readable: ReadableEvent[];
|
||||
registration(): Promise<{ connectUrl: string; release: () => void }>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Test-only observation of one control plane: every wrapped method delegates
|
||||
* unchanged. It records when each committed event becomes readable and counts
|
||||
* state reads, so a test can tell an event-driven reader from a polling one.
|
||||
*
|
||||
* `external` selects the production persistence mode, where a non-streaming
|
||||
* event stays hidden from readers until its checkpoint is saved.
|
||||
*/
|
||||
export function probeTransportPump(
|
||||
core: DurablePrpControlPlane,
|
||||
persistence: "file" | "external",
|
||||
onReadable?: (event: ReadableEvent) => void,
|
||||
): TransportPumpProbe {
|
||||
const readable: ReadableEvent[] = [];
|
||||
let lastSeq = 0;
|
||||
let reads = 0;
|
||||
let observing = false;
|
||||
const observe = () => {
|
||||
observing = true;
|
||||
try {
|
||||
for (const event of core.store.state.committedEvents) {
|
||||
if (event.sourceSeq <= lastSeq) continue;
|
||||
lastSeq = event.sourceSeq;
|
||||
const seen = { sourceSeq: event.sourceSeq, eventType: event.eventType, at: performance.now() };
|
||||
readable.push(seen);
|
||||
onReadable?.(seen);
|
||||
}
|
||||
} finally {
|
||||
observing = false;
|
||||
}
|
||||
};
|
||||
|
||||
// The working store exists before external persistence replaces the public one.
|
||||
const working = core.store as unknown as Record<"commit" | "commitEvent" | "save", (...args: unknown[]) => unknown>;
|
||||
for (const method of ["commit", "save"] as const) {
|
||||
const original = working[method].bind(working);
|
||||
working[method] = (...args: unknown[]) => {
|
||||
const result = original(...args);
|
||||
observe();
|
||||
return result;
|
||||
};
|
||||
}
|
||||
// A staged event write publishes its state partway through, right after it
|
||||
// asks whether the sender still owns the session. Look once that synchronous
|
||||
// step ends, which is before any timer or I/O callback can read the event.
|
||||
const commitEvent = working.commitEvent.bind(working);
|
||||
working.commitEvent = (build: unknown, isCurrentOwner: unknown, ...rest: unknown[]) =>
|
||||
(commitEvent(build, () => {
|
||||
queueMicrotask(observe);
|
||||
return (isCurrentOwner as () => boolean)();
|
||||
}, ...rest) as Promise<unknown>).finally(observe);
|
||||
if (persistence === "external") {
|
||||
core.setCheckpointPersistence(async () => undefined);
|
||||
const flush = core.flushCheckpoint.bind(core);
|
||||
core.flushCheckpoint = () => {
|
||||
const saved = flush();
|
||||
void saved.then(observe, () => undefined);
|
||||
return saved;
|
||||
};
|
||||
}
|
||||
|
||||
const store = core.store;
|
||||
let owner: object | null = store;
|
||||
let descriptor: PropertyDescriptor | undefined;
|
||||
while (owner !== null && descriptor === undefined) {
|
||||
descriptor = Object.getOwnPropertyDescriptor(owner, "state");
|
||||
owner = Object.getPrototypeOf(owner);
|
||||
}
|
||||
const readState = descriptor!.get!;
|
||||
Object.defineProperty(store, "state", {
|
||||
configurable: true,
|
||||
get() {
|
||||
if (!observing) reads += 1;
|
||||
return readState.call(store);
|
||||
},
|
||||
});
|
||||
|
||||
return {
|
||||
stateReads: () => reads,
|
||||
readable,
|
||||
async registration() {
|
||||
await core.start();
|
||||
return { connectUrl: core.connectUrl, release: () => undefined };
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in new issue
Block a user