1import * as NodeModule from "node:module";
2import * as NodeURL from "node:url";
3import * as NodeCrypto from "node:crypto";
4import * as NodeAssert from "node:assert/strict";
5const root = process.argv[2];
6const mode = process.argv[3];
7const withHandoff = mode.startsWith("handoff");
8const require = NodeModule.createRequire(root + "/apps/server/package.json");
9const load = (name) => import(NodeURL.pathToFileURL(require.resolve("effect/" + name)));
10const [Effect, Layer, FileSystem, Path] = await Promise.all([
11 load("Effect"),
12 load("Layer"),
13 load("FileSystem"),
14 load("Path"),
15]);
16const app = (file) => import(NodeURL.pathToFileURL(root + "/apps/server/src/" + file + ".ts"));
17const providerCore = (file) =>
18 import(NodeURL.pathToFileURL(root + "/packages/provider-core/src/server/" + file + ".ts"));
19const [Start, Projection, Run, Sessions, Policy, Id, Sink, Handoff, Git, Project, Auth] =
20 await Promise.all([
21 app("orchestration-v2/ProviderTurnStartService"),
22 app("orchestration-v2/ProjectionStore"),
23 app("orchestration-v2/RunExecutionService"),
24 app("orchestration-v2/ProviderSessionManager"),
25 app("orchestration-v2/RuntimePolicy"),
26 providerCore("IdAllocator"),
27 app("orchestration-v2/EventSink"),
28 app("orchestration-v2/ContextHandoffService"),
29 app("git/GitWorkflowService"),
30 app("project/ProjectService"),
31 app("provider/ProviderAuthService"),
32 ]);
33let current;
34let fullReads = 0;
35let threadReads = 0;
36const liveRuns = [];
37const refs = [];
38const checkpoints = [];
39const count = Number(process.argv[4] ?? 4);
40const bytes = Number(process.argv[5] ?? 32768);
41const session = {
42 driver: "codex",
43 providerSession: { id: "session", driver: "codex" },
44 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
45 startTurn: () =>
46 mode === "handoff-failure" ? Effect.fail("Synthetic startup failure") : Effect.void,
47 compactThread: () => Effect.void,
48};
49const dependencies = Layer.mergeAll(
50 Layer.mock(Handoff.ContextHandoffServiceV2)({}),
51 Id.layer,
52 FileSystem.layerNoop({}),
53 Path.layer,
54 Layer.mock(Git.GitWorkflowService)({}),
55 Layer.mock(Project.ProjectService)({}),
56 Layer.mock(Auth.ProviderAuthService)({}),
57 Layer.mock(Projection.ProjectionStoreV2)({
58 getThreadProjection: () =>
59 Effect.sync(() => {
60 fullReads++;
61 throw new Error("full transcript read");
62 }),
63 hasUnpairedRunInterruptRequest: () => Effect.succeed(false),
64 getThreadRecords: (threadId, fields) =>
65 Effect.sync(() => {
66 NodeAssert.equal(threadId, current.thread.id);
67 NodeAssert.deepEqual(fields, []);
68 threadReads++;
69 return { thread: current.thread };
70 }),
71 getTurnStartContext: () =>
72 Effect.sync(() => {
73 fullReads++;
74 return { ...current, hasConversation: true };
75 }),
76 getTurnStartHistory: () => Effect.sync(() => current.turnItems),
77 getRuntimeRecoveryProjection: () => Effect.sync(() => current),
78 }),
79 Layer.mock(Sessions.ProviderSessionManagerV2)({ open: () => Effect.succeed(session) }),
80 Layer.mock(Policy.RuntimePolicyV2)({
81 resolve: () =>
82 Effect.succeed({ cwd: "/synthetic", interactionMode: "default", runtimeMode: "full-access" }),
83 }),
84 Layer.mock(Run.RunExecutionServiceV2)({
85 startRootRun: (input) =>
86 Effect.gen(function* () {
87 liveRuns.push(input);
88 const turnInput = { ...input, message: input.message };
89 const result = yield* Effect.exit(
90 mode === "handoff-compact"
91 ? input.session.compactThread(turnInput)
92 : input.session.startTurn(turnInput),
93 );
94 NodeAssert.equal(result._tag, mode === "handoff-failure" ? "Failure" : "Success");
95 }),
96 }),
97 Layer.mock(Sink.EventSinkV2)({
98 write: () => Effect.succeed([]),
99 writeIfRunCurrent: ({ events }) =>
100 Effect.sync(() => {
101 for (const event of events)
102 if (event.type === "run.updated") current.runs = [event.payload];
103 return { committed: true, storedEvents: [] };
104 }),
105 }),
106);
107function fixture(index) {
108 const runId = "run:" + index;
109 const history = JSON.parse(
110 JSON.stringify({
111 id: "historical:" + index,
112 threadId: "thread",
113 runId: null,
114 providerThreadId: "provider",
115 status: "completed",
116 type: "command_execution",
117 output: NodeCrypto.randomBytes(bytes).toString("hex"),
118 }),
119 );
120 refs.push(new WeakRef(history));
121 return {
122 thread: { id: "thread", projectId: "project", branch: null, worktreePath: null },
123 runs: [
124 {
125 id: runId,
126 threadId: "thread",
127 ordinal: 1,
128 providerInstanceId: "instance",
129 modelSelection: { instanceId: "instance", model: "synthetic" },
130 providerThreadId: "provider",
131 userMessageId: "message",
132 rootNodeId: "root",
133 activeAttemptId: "attempt",
134 status: "starting",
135 startedAt: null,
136 },
137 ],
138 nodes: [{ id: "root", checkpointScopeId: "scope" }],
139 attempts: [{ id: "attempt", runId, providerThreadId: "provider" }],
140 providerThreads: [
141 {
142 id: "provider",
143 driver: "codex",
144 providerSessionId: "session",
145 nativeThreadRef: { nativeId: "native", driver: "codex", strength: "strong" },
146 contextUsage: { usedTokens: 0, maxTokens: 128000 },
147 handoffIds: [],
148 },
149 ],
150 messages: [{ id: "message", text: "Continue", attachments: [] }],
151 checkpointScopes: [{ id: "scope" }],
152 contextHandoffs: withHandoff
153 ? [
154 {
155 id: "handoff",
156 threadId: "thread",
157 targetRunId: runId,
158 toProviderThreadId: "provider",
159 fromProviderThreadIds: ["provider"],
160 coveredRunOrdinals: { from: 1, to: 1 },
161 status: "ready",
162 strategy: "full_thread_summary",
163 summaryText: "Synthetic summary",
164 handoffIds: [],
165 },
166 ]
167 : [],
168 contextTransfers: [],
169 providerSessions: [],
170 providerTurns: [],
171 subagents: [],
172 turnItems: [history],
173 };
174}
175await Effect.runPromise(
176 Effect.gen(function* () {
177 const service = yield* Start.ProviderTurnStartServiceV2;
178 for (let i = 0; i < count; i++) {
179 current = fixture(i);
180 yield* service.start({ threadId: "thread", runId: "run:" + i });
181 // The run worker must keep checking current durable state, not its startup copy.
182 const controls = liveRuns.at(-1);
183 current = { ...current, turnItems: [] };
184 NodeAssert.equal(yield* controls.shouldStartProviderTurn(), true);
185 NodeAssert.equal(yield* controls.shouldFinalizeRun(), true);
186 const liveRun = current.runs[0];
187 current = { ...current, runs: [{ ...liveRun, activeAttemptId: "replacement" }] };
188 NodeAssert.equal(yield* controls.shouldStartProviderTurn(), false);
189 NodeAssert.equal(yield* controls.shouldFinalizeRun(), false);
190 current = { ...current, runs: [{ ...liveRun, status: "completed" }] };
191 NodeAssert.equal(yield* controls.shouldFinalizeRun(), false);
192 NodeAssert.deepEqual(yield* controls.loadInheritedBackgroundTurnItems(), []);
193 NodeAssert.equal(fullReads, i + 1);
194 NodeAssert.equal(threadReads, i + 1);
195 NodeAssert.equal(yield* controls.hasUnpairedRunInterruptRequest(), false);
196 NodeAssert.equal(fullReads, i + 1);
197 NodeAssert.equal(threadReads, i + 1);
198 current = null;
199 if (i === Math.floor(count / 2) - 1 || i === count - 1) {
200 yield* Effect.promise(async () => {
201 for (let j = 0; j < 3; j++) {
202 await new Promise(setImmediate);
203 global.gc();
204 }
205 checkpoints.push({
206 runs: liveRuns.length,
207 payloadMiB: ((i + 1) * bytes * 2) / 1048576,
208 retained: refs.filter((ref) => ref.deref()).length,
209 heapMiB: Math.round(process.memoryUsage().heapUsed / 1048576),
210 });
211 });
212 }
213 }
214 }).pipe(Effect.provide(Start.layer.pipe(Layer.provide(dependencies)))),
215);
216console.log(JSON.stringify({ checkpoints }));