1/**
2 * OpenCode 2 through the whole orchestrator, against a replayed HTTP server:
3 * the transcript fixes the order of every request the adapter sends, so a
4 * request the orchestrator never lets it make fails the run.
5 */
6import * as NodeServices from "@effect/platform-node/NodeServices";
7import { assert, describe, it } from "@effect/vitest";
8import {
9 CommandId,
10 MessageId,
11 type ModelSelection,
12 type OrchestrationV2Command,
13 type OrchestrationV2Run,
14 ProjectId,
15 type ProviderInteractionMode,
16 ProviderInstanceId,
17 type ProviderReplayEntry,
18 type RuntimeMode,
19 ThreadId,
20} from "@t3tools/contracts";
21import * as Effect from "effect/Effect";
22import * as Layer from "effect/Layer";
23import * as FileSystem from "effect/FileSystem";
24import * as Path from "effect/Path";
25
26import {
27 OPENCODE2_HTTP_PROTOCOL,
28 OpenCode2OrchestratorReplayHarness,
29} from "./Adapters/OpenCode2AdapterV2.testkit.ts";
30import { OPENCODE_PROVIDER } from "@t3tools/provider-opencode/testing";
31import { provideDeterministicTestRuntime } from "./testkit/DeterministicRuntime.ts";
32import type { OrchestratorV2ScenarioStep } from "./testkit/OrchestratorScenario.ts";
33import { runOrchestratorV2ProviderReplayScenario } from "./testkit/ProviderReplayHarness.ts";
34import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
35import {
36 decodeProviderReplayNdjson,
37 readProviderReplayTranscript,
38} from "@t3tools/provider-testing/replayTranscript";
39import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
40
41const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX";
42/** Held until the scenario releases it, so the turn is still running meanwhile. */
43const FIRST_TURN_END = "first-turn-end";
44/** Held until the scenario releases it, so a follow-up is running when its stream drops. */
45const BACKGROUND_HOLD = "background-hold";
46/** The parent session of the recorded background run. */
47const BACKGROUND_PARENT = "ses_f1485cda4ffeXuc6GjAU9vDiRb";
48/** Its background subagent's session. */
49const BACKGROUND_CHILD = "ses_f1485c529ffea4URrYruwEg0Ja";
50const BACKGROUND_PROMPT =
51 "Use the subagent tool with background enabled to delegate to the general subagent with the prompt: 'Run the shell command `sleep 20` with the bash tool and then reply exactly CHILD_OK.' As soon as it is launched, reply exactly PARENT_OK and end your turn without waiting for it.";
52const instanceId = ProviderInstanceId.make("opencode");
53const bigPickle: ModelSelection = { instanceId, model: "opencode/big-pickle" };
54const mimo: ModelSelection = { instanceId, model: "opencode/mimo-v2.6-flash-free" };
55const nemotron: ModelSelection = { instanceId, model: "opencode/nemotron-3.5-lightning-free" };
56
57const out = (type: string, input?: unknown): ProviderReplayEntry => ({
58 type: "expect_outbound",
59 frame: input === undefined ? { type } : { type, input },
60});
61const reply = (operation: string, data: unknown): ProviderReplayEntry => ({
62 type: "emit_inbound",
63 frame: { type: "sdk.response", operation, data },
64});
65const event = (type: string, data: Record<string, unknown>): ProviderReplayEntry => ({
66 type: "emit_inbound",
67 frame: {
68 type: "sdk.event",
69 event: { id: `evt_${type.replaceAll(".", "")}`, created: 1, type, data },
70 },
71});
72const labelled = (entry: ProviderReplayEntry, label: string): ProviderReplayEntry =>
73 entry.type === "runtime_exit" ? entry : { ...entry, label };
74/**
75 * T3's own rules after a mode's: every thread's T3 MCP server is denied, and
76 * then this thread's own is allowed again (last match wins).
77 */
78const mcpRules = (name: string) => [
79 { action: "t3-code-*", resource: "*", effect: "deny" },
80 { action: `t3-code-thread_${name}_*`, resource: "*", effect: "allow" },
81];
82const FULL_ACCESS = [{ action: "*", resource: "*", effect: "allow" }];
83/** Full access for the thread named `name`. */
84const t3Rules = (name: string) => [...FULL_ACCESS, ...mcpRules(name)];
85/** Paths the build and plan agents allow for themselves, as 2.0.18 lists them. */
86const BUILD_PATHS = [
87 {
88 action: "external_directory",
89 resource: "/home/.local/share/opencode/tool-output/*",
90 effect: "allow",
91 },
92];
93const PLAN_PATHS = [
94 ...BUILD_PATHS,
95 { action: "edit", resource: "/home/.opencode/plan/*", effect: "allow" },
96 { action: "external_directory", resource: "/home/.opencode/plan/*", effect: "allow" },
97];
98const agentInfo = (id: string, description: string, permissions: ReadonlyArray<unknown>) => ({
99 id,
100 name: id === "plan" ? "Plan" : "Build",
101 request: { settings: {}, headers: {}, body: {} },
102 description,
103 mode: "primary",
104 hidden: false,
105 permissions,
106});
107/** `/api/agent` trimmed to the two agents a T3 session runs. */
108const agentList = (directory: string) => ({
109 location: { directory },
110 data: [
111 agentInfo("build", "The default agent.", [
112 { action: "*", resource: "*", effect: "allow" },
113 { action: "external_directory", resource: "*", effect: "ask" },
114 ...BUILD_PATHS,
115 ]),
116 agentInfo("plan", "Read-only agent for planning.", [
117 { action: "*", resource: "*", effect: "allow" },
118 { action: "external_directory", resource: "*", effect: "ask" },
119 ...BUILD_PATHS,
120 { action: "edit", resource: "*", effect: "deny" },
121 ...PLAN_PATHS.slice(BUILD_PATHS.length),
122 ]),
123 ],
124});
125/** A session this runtime loads again (after a detach) waits on nothing. */
126const noOpenRequests: ReadonlyArray<ProviderReplayEntry> = [
127 out("permission.list", { sessionID: SESSION }),
128 reply("permission.list", { data: [] }),
129 out("session.form.list", { sessionID: SESSION }),
130 reply("session.form.list", { data: [] }),
131];
132const supervisedRules = (name: string) => [
133 { action: "shell", resource: "*", effect: "ask" },
134 { action: "edit", resource: "*", effect: "ask" },
135 { action: "external_directory", resource: "*", effect: "ask" },
136 ...BUILD_PATHS,
137 ...mcpRules(name),
138];
139const autoEditRules = (name: string) => [
140 { action: "shell", resource: "*", effect: "ask" },
141 { action: "edit", resource: "*", effect: "allow" },
142 { action: "external_directory", resource: "*", effect: "ask" },
143 ...BUILD_PATHS,
144 ...mcpRules(name),
145];
146/** Plan mode on Full access: edits are denied except the plan agent's own plan files. */
147const planRules = (name: string) => [
148 { action: "*", resource: "*", effect: "allow" },
149 { action: "edit", resource: "*", effect: "deny" },
150 ...PLAN_PATHS,
151 ...mcpRules(name),
152];
153
154const sessionInfo = (directory: string, permissions: ReadonlyArray<unknown>, title: string) => ({
155 data: {
156 id: SESSION,
157 title,
158 projectID: "global",
159 model: { id: "big-pickle", providerID: "opencode", variant: "default" },
160 cost: 0,
161 tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
162 time: { created: 1790656601394, updated: 1790656601394 },
163 location: { directory },
164 permissions,
165 },
166});
167/** T3's instructions entry, written before a thread's first prompt and whenever it changes. */
168const instructionsWritten: ReadonlyArray<ProviderReplayEntry> = [
169 out("session.instructions.entry.put", { sessionID: SESSION, key: "t3-code", value: "<any>" }),
170 reply("session.instructions.entry.put", null),
171];
172/** One prompt the server accepts and answers with `text`. */
173const answeredPrompt = (text: string): ReadonlyArray<ProviderReplayEntry> => [
174 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
175 reply("session.prompt", {
176 data: {
177 id: `msg_user_${text}`,
178 sessionID: SESSION,
179 time: { created: 1790656601410 },
180 type: "user",
181 payload: { text: "<prompt>" },
182 delivery: "steer",
183 },
184 }),
185 event("session.text.ended", {
186 sessionID: SESSION,
187 assistantMessageID: `msg_assistant_${text}`,
188 ordinal: 0,
189 text,
190 }),
191 event("session.execution.succeeded", { sessionID: SESSION }),
192];
193/** The model list read the first time a thread runs in `directory`. */
194const directoryModels = (directory: string): ReadonlyArray<ProviderReplayEntry> => [
195 out("model.list", { "location[directory]": directory }),
196 reply("model.list", {
197 location: { directory },
198 data: [
199 catalogModel("big-pickle", "Big Pickle"),
200 catalogModel("mimo-v2.6-flash-free", "MiMo V2.6 Flash Free"),
201 ],
202 }),
203];
204/** A `/api/model` entry as 2.0.18 lists it; a known window means no re-read after a turn. */
205const catalogModel = (id: string, name: string) => ({
206 id,
207 modelID: id,
208 providerID: "opencode",
209 family: id,
210 name,
211 compatibility: { reasoningField: "reasoning_content" },
212 package: "@opencode/ai/providers/openai-compatible",
213 settings: { apiKey: "public", baseURL: "https://opencode.ai/zen/v1", provider: "opencode" },
214 capabilities: { tools: true, input: ["text"], output: ["text"] },
215 variants: [],
216 time: { released: 1760659200000 },
217 cost: [{ input: 0, output: 0, cache: { read: 0, write: 0 } }],
218 status: "active",
219 enabled: true,
220 limit: { context: 200000, input: 160000, output: 32000 },
221});
222const createdSession = (
223 directory: string,
224 name: string,
225 permissions: ReadonlyArray<unknown> = t3Rules(name),
226 // Only a mode that narrows Full access reads the agents' own path rules.
227 narrows = false,
228): ReadonlyArray<ProviderReplayEntry> => [
229 out("event.subscribe"),
230 out("model.list", "<any>"),
231 reply("model.list", {
232 location: { directory },
233 data: [
234 catalogModel("big-pickle", "Big Pickle"),
235 catalogModel("mimo-v2.6-flash-free", "MiMo V2.6 Flash Free"),
236 ],
237 }),
238 ...(narrows ? [out("agent.list", "<any>"), reply("agent.list", agentList(directory))] : []),
239 // The thread's title, which is its name here.
240 out("session.create", { location: { directory }, model: "<any>", permissions, title: name }),
241 reply("session.create", sessionInfo(directory, permissions, name)),
242];
243
244/** A recording with its scrubbed `<work>` directory replaced by `directory`. */
245const withDirectory = <T>(value: T, directory: string): T => {
246 const replace = (entry: unknown): unknown =>
247 entry === "<work>"
248 ? directory
249 : Array.isArray(entry)
250 ? entry.map(replace)
251 : typeof entry === "object" && entry !== null
252 ? Object.fromEntries(Object.entries(entry).map(([key, inner]) => [key, replace(inner)]))
253 : entry;
254 return replace(value) as T;
255};
256
257const threadCommands = (input: {
258 readonly name: string;
259 readonly worktreePath: string;
260 readonly runtimeMode?: RuntimeMode;
261 readonly interactionMode?: ProviderInteractionMode;
262}) => {
263 const threadId = ThreadId.make(`thread:${input.name}`);
264 const command = (key: string) => CommandId.make(`command:${input.name}:${key}`);
265 return {
266 threadId,
267 create: {
268 type: "thread.create",
269 createdBy: "user",
270 creationSource: "web",
271 commandId: command("create"),
272 threadId,
273 projectId: ProjectId.make(`project:${input.name}`),
274 title: input.name,
275 modelSelection: bigPickle,
276 runtimeMode: input.runtimeMode ?? "full-access",
277 interactionMode: input.interactionMode ?? "default",
278 branch: null,
279 worktreePath: input.worktreePath,
280 } satisfies OrchestrationV2Command,
281 message: (key: string, modelSelection: ModelSelection = bigPickle, text?: string) =>
282 ({
283 type: "message.dispatch",
284 createdBy: "user",
285 creationSource: "web",
286 commandId: command(key),
287 threadId,
288 messageId: MessageId.make(`message:${input.name}:${key}`),
289 text: text ?? `Reply with exactly: ${key}`,
290 attachments: [],
291 modelSelection,
292 dispatchMode: { type: "start_immediately" },
293 }) satisfies OrchestrationV2Command,
294 interactionMode: (key: string, interactionMode: "default" | "plan") =>
295 ({
296 type: "thread.interaction-mode.set",
297 commandId: command(key),
298 threadId,
299 interactionMode,
300 }) satisfies OrchestrationV2Command,
301 command,
302 };
303};
304
305/** Runs `commands` in order, letting the thread go idle after each message. */
306const runScenario = (input: {
307 readonly name: string;
308 readonly threadId: ThreadId;
309 readonly entries: ReadonlyArray<ProviderReplayEntry>;
310 readonly commands: ReadonlyArray<OrchestrationV2Command>;
311}) =>
312 Effect.gen(function* () {
313 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
314 provider: OPENCODE_PROVIDER,
315 protocol: OPENCODE2_HTTP_PROTOCOL,
316 version: "2.0.18",
317 scenario: input.name,
318 entries: input.entries,
319 });
320 const steps = input.commands.flatMap((command): Array<OrchestratorV2ScenarioStep> => [
321 { type: "dispatch", command },
322 { type: "advance_clock", duration: "1 millis" },
323 ...(command.type === "message.dispatch"
324 ? [{ type: "await_thread_idle" as const, threadId: input.threadId }]
325 : []),
326 ]);
327 const result = yield* runOrchestratorV2ProviderReplayScenario(
328 { name: input.name, transcript, commands: input.commands, steps },
329 OpenCode2OrchestratorReplayHarness,
330 ).pipe(provideDeterministicTestRuntime);
331 const projection = result.projections.get(input.threadId);
332 assert.isDefined(projection);
333 return projection;
334 });
335
336describe("OpenCode 2 through the orchestrator", () => {
337 it.effect.each(["message", "thread settings"] as const)(
338 "switches the session's model before the next prompt when changed from the %s",
339 (via) =>
340 Effect.gen(function* () {
341 const name = `opencode2-model-switch-${via.replace(" ", "-")}`;
342 const cwd = yield* checkpointWorkspace(name);
343 const thread = threadCommands({ name, worktreePath: cwd });
344 const projection = yield* runScenario({
345 name,
346 threadId: thread.threadId,
347 entries: [
348 ...createdSession(cwd, name),
349 ...instructionsWritten,
350 ...answeredPrompt("FIRST"),
351 // The next turn resumes the session at its new selection.
352 out("session.get", { sessionID: SESSION }),
353 reply("session.get", sessionInfo(cwd, t3Rules(name), name)),
354 out("session.switchModel", {
355 sessionID: SESSION,
356 model: { providerID: "opencode", id: "mimo-v2.6-flash-free" },
357 }),
358 reply("session.switchModel", null),
359 // The instructions name the model, so they are written again.
360 ...instructionsWritten,
361 ...answeredPrompt("SECOND"),
362 ],
363 commands: [
364 thread.create,
365 thread.message("first"),
366 ...(via === "thread settings"
367 ? [
368 {
369 type: "thread.model-selection.set",
370 commandId: thread.command("model"),
371 threadId: thread.threadId,
372 modelSelection: mimo,
373 } satisfies OrchestrationV2Command,
374 ]
375 : []),
376 thread.message("second", mimo),
377 ],
378 });
379 assert.deepEqual(
380 projection.runs.map((run) => [run.status, run.modelSelection.model]),
381 [
382 ["completed", bigPickle.model],
383 ["completed", mimo.model],
384 ],
385 );
386 // One native session carried both turns.
387 assert.lengthOf(projection.providerThreads, 1);
388 }).pipe(Effect.scoped),
389 );
390
391 it.effect("moves the session to the thread's new worktree before the next prompt", () =>
392 Effect.gen(function* () {
393 const name = "opencode2-worktree-move";
394 const before = yield* checkpointWorkspace(`${name}-before`);
395 const after = yield* checkpointWorkspace(`${name}-after`);
396 const thread = threadCommands({ name, worktreePath: before });
397 const projection = yield* runScenario({
398 name,
399 threadId: thread.threadId,
400 entries: [
401 ...createdSession(before, name),
402 ...instructionsWritten,
403 ...answeredPrompt("FIRST"),
404 ...directoryModels(after),
405 out("session.get", { sessionID: SESSION }),
406 reply("session.get", sessionInfo(before, t3Rules(name), name)),
407 // The worktree change detached the thread, so its session is loaded afresh.
408 ...noOpenRequests,
409 out("session.move", { sessionID: SESSION, directory: after }),
410 reply("session.move", null),
411 // The moved thread reopens its session, which writes the entry again.
412 ...instructionsWritten,
413 ...answeredPrompt("SECOND"),
414 ],
415 commands: [
416 thread.create,
417 thread.message("first"),
418 {
419 type: "thread.metadata.update",
420 commandId: thread.command("worktree"),
421 threadId: thread.threadId,
422 worktreePath: after,
423 },
424 thread.message("second"),
425 ],
426 });
427 assert.deepEqual(
428 projection.runs.map((run) => run.status),
429 ["completed", "completed"],
430 );
431 assert.lengthOf(projection.providerThreads, 1);
432 }).pipe(Effect.scoped),
433 );
434
435 it.effect("gives a session made with older rules T3's rules before its next prompt", () =>
436 Effect.gen(function* () {
437 const name = "opencode2-resume-rules";
438 const before = yield* checkpointWorkspace(`${name}-before`);
439 const after = yield* checkpointWorkspace(`${name}-after`);
440 const thread = threadCommands({ name, worktreePath: before });
441 const projection = yield* runScenario({
442 name,
443 threadId: thread.threadId,
444 entries: [
445 ...createdSession(before, name),
446 ...instructionsWritten,
447 ...answeredPrompt("FIRST"),
448 ...directoryModels(after),
449 // Reopened after a worktree change, the session reports the rules an
450 // older build gave it, which denied subagents; they are replaced
451 // before anything runs.
452 out("session.get", { sessionID: SESSION }),
453 reply(
454 "session.get",
455 sessionInfo(
456 before,
457 [
458 { action: "*", resource: "*", effect: "allow" },
459 { action: "subagent", resource: "*", effect: "deny" },
460 ],
461 name,
462 ),
463 ),
464 ...noOpenRequests,
465 out("session.update", { sessionID: SESSION, permissions: t3Rules(name) }),
466 reply("session.update", null),
467 out("session.move", { sessionID: SESSION, directory: after }),
468 reply("session.move", null),
469 ...instructionsWritten,
470 ...answeredPrompt("SECOND"),
471 ],
472 commands: [
473 thread.create,
474 thread.message("first"),
475 {
476 type: "thread.metadata.update",
477 commandId: thread.command("worktree"),
478 threadId: thread.threadId,
479 worktreePath: after,
480 },
481 thread.message("second"),
482 ],
483 });
484 assert.deepEqual(
485 projection.runs.map((run) => run.status),
486 ["completed", "completed"],
487 );
488 }).pipe(Effect.scoped),
489 );
490
491 it.effect(
492 "creates a Supervised thread's session with rules that ask before shell and edits",
493 () =>
494 Effect.gen(function* () {
495 const name = "opencode2-supervised-rules";
496 const cwd = yield* checkpointWorkspace(name);
497 const thread = threadCommands({
498 name,
499 worktreePath: cwd,
500 runtimeMode: "approval-required",
501 });
502 const projection = yield* runScenario({
503 name,
504 threadId: thread.threadId,
505 entries: [
506 ...createdSession(cwd, name, supervisedRules(name), true),
507 ...instructionsWritten,
508 ...answeredPrompt("FIRST"),
509 ],
510 commands: [thread.create, thread.message("first")],
511 });
512 assert.deepEqual(
513 projection.runs.map((run) => run.status),
514 ["completed"],
515 );
516 }).pipe(Effect.scoped),
517 );
518
519 it.effect("rewrites the session's rules when the thread's mode changes between turns", () =>
520 Effect.gen(function* () {
521 const name = "opencode2-mode-change";
522 const cwd = yield* checkpointWorkspace(name);
523 const thread = threadCommands({ name, worktreePath: cwd });
524 const setMode = (key: string, runtimeMode: RuntimeMode) =>
525 ({
526 type: "thread.runtime-mode.set",
527 commandId: thread.command(key),
528 threadId: thread.threadId,
529 runtimeMode,
530 }) satisfies OrchestrationV2Command;
531 const projection = yield* runScenario({
532 name,
533 threadId: thread.threadId,
534 entries: [
535 ...createdSession(cwd, name),
536 ...instructionsWritten,
537 ...answeredPrompt("FIRST"),
538 // A mode change detaches nothing: the same session is resumed with the new rules.
539 out("session.get", { sessionID: SESSION }),
540 reply("session.get", sessionInfo(cwd, t3Rules(name), name)),
541 out("agent.list", "<any>"),
542 reply("agent.list", agentList(cwd)),
543 out("session.update", { sessionID: SESSION, permissions: autoEditRules(name) }),
544 reply("session.update", null),
545 ...answeredPrompt("SECOND"),
546 // Back to Full access: the narrowing rules go.
547 out("session.get", { sessionID: SESSION }),
548 reply("session.get", sessionInfo(cwd, autoEditRules(name), name)),
549 out("session.update", { sessionID: SESSION, permissions: t3Rules(name) }),
550 reply("session.update", null),
551 ...answeredPrompt("THIRD"),
552 ],
553 commands: [
554 thread.create,
555 thread.message("first"),
556 setMode("auto-edit", "auto-accept-edits"),
557 thread.message("second"),
558 setMode("full", "full-access"),
559 thread.message("third"),
560 ],
561 });
562 assert.deepEqual(
563 projection.runs.map((run) => run.status),
564 ["completed", "completed", "completed"],
565 );
566 assert.lengthOf(projection.providerThreads, 1);
567 }).pipe(Effect.scoped),
568 );
569
570 it.effect("denies edits outside the plan directory in plan mode and lifts it after", () =>
571 Effect.gen(function* () {
572 const name = "opencode2-plan-rules";
573 const cwd = yield* checkpointWorkspace(name);
574 const thread = threadCommands({ name, worktreePath: cwd, interactionMode: "plan" });
575 const projection = yield* runScenario({
576 name,
577 threadId: thread.threadId,
578 entries: [
579 ...createdSession(cwd, name, planRules(name), true),
580 // Plan mode is also OpenCode's plan agent, switched before the prompt.
581 out("session.switchAgent", { sessionID: SESSION, agent: "plan" }),
582 reply("session.switchAgent", null),
583 ...instructionsWritten,
584 ...answeredPrompt("PLANNED"),
585 out("session.get", { sessionID: SESSION }),
586 reply("session.get", sessionInfo(cwd, planRules(name), name)),
587 out("session.update", { sessionID: SESSION, permissions: t3Rules(name) }),
588 reply("session.update", null),
589 out("session.switchAgent", { sessionID: SESSION, agent: "build" }),
590 reply("session.switchAgent", null),
591 ...answeredPrompt("BUILT"),
592 ],
593 commands: [
594 thread.create,
595 thread.message("plan"),
596 {
597 type: "thread.interaction-mode.set",
598 commandId: thread.command("default"),
599 threadId: thread.threadId,
600 interactionMode: "default",
601 },
602 thread.message("build"),
603 ],
604 });
605 assert.deepEqual(
606 projection.runs.map((run) => run.status),
607 ["completed", "completed"],
608 );
609 }).pipe(Effect.scoped),
610 );
611
612 it.effect("forks from an earlier turn before the next turn's user message", () =>
613 Effect.gen(function* () {
614 const name = "opencode2-fork";
615 const cwd = yield* checkpointWorkspace(name);
616 const recorded = yield* readProviderReplayTranscript(
617 new URL("./testkit/fixtures/opencode2_fork/opencode_transcript.ndjson", import.meta.url),
618 );
619 // The recording scrubbed its directory to `<work>`; the fork it answers
620 // runs where its source does, which is this test's workspace.
621 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript(
622 withDirectory(recorded, cwd),
623 );
624 const source = threadCommands({ name, worktreePath: cwd });
625 const target = ThreadId.make(`thread:${name}:target`);
626 const [one, two] = [source.message("one"), source.message("two")];
627 const commands: ReadonlyArray<OrchestrationV2Command> = [
628 source.create,
629 one,
630 two,
631 {
632 type: "thread.fork",
633 createdBy: "user",
634 creationSource: "web",
635 commandId: source.command("fork"),
636 sourceThreadId: source.threadId,
637 targetThreadId: target,
638 sourcePoint: {
639 type: "run",
640 runId: (yield* IdAllocator.IdAllocatorV2).derive.run({
641 threadId: source.threadId,
642 ordinal: 1,
643 }),
644 },
645 },
646 {
647 ...source.message("repeat"),
648 threadId: target,
649 messageId: MessageId.make(`message:${name}:repeat`),
650 },
651 ];
652 const steps: Array<OrchestratorV2ScenarioStep> = commands.flatMap((command) => [
653 { type: "dispatch" as const, command },
654 { type: "advance_clock" as const, duration: "1 millis" as const },
655 ...(command.type === "message.dispatch"
656 ? [{ type: "await_thread_idle" as const, threadId: command.threadId }]
657 : []),
658 ]);
659 const result = yield* runOrchestratorV2ProviderReplayScenario(
660 { name, transcript, commands, steps, projectionThreadIds: [source.threadId, target] },
661 OpenCode2OrchestratorReplayHarness,
662 ).pipe(provideDeterministicTestRuntime);
663 const forked = result.projections.get(target);
664 assert.isDefined(forked);
665 assert.equal(forked.contextTransfers[0]?.resolution?.strategy, "native_fork");
666 assert.equal(
667 forked.providerThreads[0]?.nativeThreadRef?.nativeId,
668 recorded.metadata?.["forkedNativeSessionId"],
669 );
670 // The fork keeps the first turn and drops the second: the model answers
671 // from the first alone, and T3 shows the inherited turn but not the other.
672 assert.deepEqual(
673 forked.runs.map((run) => run.status),
674 ["completed"],
675 );
676 const visible = forked.visibleTurnItems.flatMap((row) =>
677 row.item.type === "assistant_message" ? [row.item.text] : [],
678 );
679 assert.deepEqual(visible, ["ONE", "ONE"]);
680 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
681 );
682
683 it.effect("never sends OpenCode a queued message that was cancelled", () =>
684 Effect.gen(function* () {
685 const name = "opencode2-queued-cancel";
686 const cwd = yield* checkpointWorkspace(name);
687 const thread = threadCommands({ name, worktreePath: cwd });
688 const queued = (key: string): OrchestrationV2Command => ({
689 ...thread.message(key),
690 dispatchMode: { type: "queue_after_active" },
691 });
692 const ids = yield* IdAllocator.IdAllocatorV2;
693 const run = (ordinal: number) => ids.derive.run({ threadId: thread.threadId, ordinal });
694 // The first turn is held open until the queue has been changed: its end
695 // is the last event, after the cancel.
696 const [first, second, third] = [thread.message("first"), queued("second"), queued("third")];
697 const cancel: OrchestrationV2Command = {
698 type: "queued-run.cancel",
699 commandId: thread.command("cancel"),
700 threadId: thread.threadId,
701 runId: run(2),
702 };
703 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
704 provider: OPENCODE_PROVIDER,
705 protocol: OPENCODE2_HTTP_PROTOCOL,
706 version: "2.0.18",
707 scenario: name,
708 entries: [
709 ...createdSession(cwd, name),
710 ...instructionsWritten,
711 out("session.prompt", { sessionID: SESSION, id: "<any>", text: "<any>" }),
712 reply("session.prompt", {
713 data: {
714 id: "msg_user_FIRST",
715 sessionID: SESSION,
716 time: { created: 1 },
717 type: "user",
718 payload: { text: "<prompt>" },
719 delivery: "steer",
720 },
721 }),
722 // Only the third message reaches OpenCode, as the next turn.
723 ...answeredPrompt("THIRD"),
724 ],
725 });
726 const steps: Array<OrchestratorV2ScenarioStep> = [
727 { type: "dispatch", command: thread.create },
728 { type: "advance_clock", duration: "1 millis" },
729 { type: "dispatch", command: first, await: false, key: "first" },
730 { type: "await_run_steerable", threadId: thread.threadId, runId: run(1) },
731 { type: "dispatch", command: second },
732 { type: "dispatch", command: third },
733 { type: "dispatch", command: cancel },
734 { type: "await_run_status", threadId: thread.threadId, runId: run(2), status: "cancelled" },
735 { type: "release_replay_gate", label: FIRST_TURN_END },
736 { type: "await", key: "first" },
737 { type: "await_thread_idle", threadId: thread.threadId },
738 ];
739 const result = yield* runOrchestratorV2ProviderReplayScenario(
740 {
741 name,
742 transcript: {
743 ...transcript,
744 entries: [
745 ...transcript.entries.slice(0, -4),
746 labelled(
747 event("session.execution.succeeded", { sessionID: SESSION }),
748 FIRST_TURN_END,
749 ),
750 ...transcript.entries.slice(-4),
751 ],
752 },
753 commands: [thread.create, first, second, third, cancel],
754 steps,
755 },
756 OpenCode2OrchestratorReplayHarness,
757 ).pipe(provideDeterministicTestRuntime);
758 const projection = result.projections.get(thread.threadId);
759 assert.isDefined(projection);
760 assert.deepEqual(
761 projection.runs.map((candidate) => candidate.status),
762 ["completed", "cancelled", "completed"],
763 );
764 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
765 );
766
767 /**
768 * The recorded background run (`opencode2_background`): the parent's turn
769 * ends while its subagent runs, then the subagent's report wakes the parent
770 * and T3 opens a continuation run for that follow-up. `reconnect` replaces
771 * the recording from `cut` on with a stream drop, a restarted stream and
772 * what the server answers then.
773 */
774 const backgroundReconnect = Effect.fn("backgroundReconnect")(function* (input: {
775 readonly name: string;
776 readonly cut: string;
777 readonly hold?: string;
778 readonly reconnect: (parent: string) => ReadonlyArray<ProviderReplayEntry>;
779 readonly steps: (
780 thread: ReturnType<typeof threadCommands>,
781 run: (ordinal: number) => OrchestrationV2Run["id"],
782 ) => ReadonlyArray<OrchestratorV2ScenarioStep>;
783 readonly commands: (
784 thread: ReturnType<typeof threadCommands>,
785 ) => ReadonlyArray<OrchestrationV2Command>;
786 }) {
787 const cwd = yield* checkpointWorkspace(input.name);
788 const recorded = yield* readProviderReplayTranscript(
789 new URL(
790 "./testkit/fixtures/opencode2_background/opencode_transcript.ndjson",
791 import.meta.url,
792 ),
793 );
794 const cut = recorded.entries.findIndex(
795 (entry) => entry.type !== "runtime_exit" && entry.label === input.cut,
796 );
797 assert.isAtLeast(cut, 0);
798 const kept = recorded.entries
799 .slice(0, cut)
800 .map((entry) =>
801 entry.type !== "runtime_exit" && entry.label === input.hold
802 ? labelled(entry, BACKGROUND_HOLD)
803 : entry,
804 );
805 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
806 ...recorded,
807 scenario: input.name,
808 entries: [
809 ...kept,
810 { type: "runtime_exit", status: "success" },
811 ...input.reconnect(BACKGROUND_PARENT),
812 ],
813 });
814 const thread = threadCommands({ name: input.name, worktreePath: cwd });
815 const ids = yield* IdAllocator.IdAllocatorV2;
816 const run = (ordinal: number) => ids.derive.run({ threadId: thread.threadId, ordinal });
817 const result = yield* runOrchestratorV2ProviderReplayScenario(
818 {
819 name: input.name,
820 transcript,
821 commands: input.commands(thread),
822 steps: input.steps(thread, run),
823 },
824 OpenCode2OrchestratorReplayHarness,
825 // Opens the continuation run a follow-up asks for, as the server does.
826 { runContinuationWorker: true },
827 ).pipe(provideDeterministicTestRuntime);
828 const projection = result.projections.get(thread.threadId);
829 assert.isDefined(projection);
830 return { result, projection };
831 });
832
833 it.effect("backfills a follow-up whose stream dropped mid-run and ends it completed", () =>
834 Effect.gen(function* () {
835 const { result, projection } = yield* backgroundReconnect({
836 name: "opencode2-follow-up-reconnect",
837 // The follow-up streams its reply; then the stream drops before the
838 // execution's end.
839 cut: "session.step.streamed.5",
840 hold: "session.text.ended.3",
841 reconnect: (parent) => [
842 out("event.subscribe"),
843 out("session.active"),
844 reply("session.active", { data: {} }),
845 // The follow-up's history since the report it answers, newest first.
846 out("message.list", { sessionID: parent, order: "desc", limit: "50" }),
847 reply("message.list", {
848 data: [
849 {
850 id: "msg_0eb7a99c0001idleFollowUp00",
851 time: { created: 5 },
852 type: "idle",
853 outcome: "succeeded",
854 },
855 {
856 id: "msg_0eb7a96bb00170K1f1HJUeBNwF",
857 time: { created: 4 },
858 type: "assistant",
859 agent: "build",
860 model: { id: "big-pickle", providerID: "opencode", variant: "default" },
861 content: [
862 {
863 type: "text",
864 text: "The background subagent finished and returned `CHILD_OK`.",
865 },
866 ],
867 finish: "stop",
868 },
869 {
870 id: "msg_0eb7a3aee0015KmTB8XkMDW8iy",
871 time: { created: 3 },
872 type: "synthetic",
873 text: "<subagent>CHILD_OK</subagent>",
874 },
875 {
876 id: "msg_0eb7a3e3c001idleParentTurn0",
877 time: { created: 2 },
878 type: "idle",
879 outcome: "succeeded",
880 },
881 ],
882 cursor: {},
883 }),
884 ],
885 commands: (thread) => [
886 thread.create,
887 thread.message("start", bigPickle, BACKGROUND_PROMPT),
888 ],
889 steps: (thread, run) => [
890 { type: "dispatch", command: thread.create },
891 { type: "advance_clock", duration: "1 millis" },
892 // A continuation run starts while the thread is busy, so this does not wait.
893 {
894 type: "dispatch",
895 command: thread.message("start", bigPickle, BACKGROUND_PROMPT),
896 await: false,
897 key: "start",
898 },
899 { type: "advance_clock", duration: "1 millis" },
900 // The follow-up's continuation turn has taken its execution when the
901 // stream drops: the held reply end is the next event.
902 { type: "await_run_steerable", threadId: thread.threadId, runId: run(2) },
903 { type: "release_replay_gate", label: BACKGROUND_HOLD },
904 {
905 type: "await_run_status",
906 threadId: thread.threadId,
907 runId: run(2),
908 status: "completed",
909 },
910 { type: "await", key: "start" },
911 { type: "await_thread_idle", threadId: thread.threadId },
912 ],
913 });
914 assert.deepEqual(
915 projection.runs.map((candidate) => candidate.status),
916 ["completed", "completed"],
917 );
918 const followUp = projection.runs[1];
919 assert.deepEqual(
920 projection.turnItems.flatMap((item) =>
921 item.runId === followUp?.id && item.type === "assistant_message" ? [item.text] : [],
922 ),
923 ["The background subagent finished and returned `CHILD_OK`."],
924 );
925 const shell = result.shellSnapshot.threads.find((row) => row.id === projection.thread.id);
926 assert.deepEqual(shell?.pendingBackgroundTasks ?? [], []);
927 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
928 );
929
930 it.effect(
931 "settles a background subagent whose end was lost with the stream, and takes the next turn",
932 () =>
933 Effect.gen(function* () {
934 const { result, projection } = yield* backgroundReconnect({
935 name: "opencode2-background-lost",
936 // The subagent runs; the stream drops before its end, its report and
937 // the follow-up it started, none of which is read back.
938 cut: "session.step.started.3",
939 reconnect: (parent) => [
940 out("event.subscribe"),
941 // Nothing runs any more: the subagent and the follow-up both ended.
942 out("session.active"),
943 reply("session.active", { data: {} }),
944 // The subagent's turn was running, so its history since its prompt
945 // is read back: its end, as the recording's later events had it.
946 out("message.list", { sessionID: BACKGROUND_CHILD, order: "desc", limit: "50" }),
947 reply("message.list", {
948 data: [
949 {
950 id: "msg_0eb7a99b0001idleChildTurn0",
951 time: { created: 4 },
952 type: "idle",
953 outcome: "succeeded",
954 },
955 {
956 id: "msg_0eb7a916c001NRJQJ5gwIT0QnU",
957 time: { created: 3 },
958 type: "assistant",
959 agent: "general",
960 model: { id: "big-pickle", providerID: "opencode", variant: "default" },
961 content: [{ type: "text", text: "CHILD_OK" }],
962 finish: "stop",
963 },
964 {
965 id: "msg_0eb7a3aea001Lw1Pav5d4akfEg",
966 time: { created: 2 },
967 type: "user",
968 text: "Run the shell command `sleep 20` with the bash tool and then reply exactly CHILD_OK.",
969 },
970 ],
971 cursor: {},
972 }),
973 out("session.prompt", { sessionID: parent, id: "<any>", text: "<any>" }),
974 reply("session.prompt", {
975 data: {
976 id: "msg_next",
977 sessionID: parent,
978 time: { created: 9 },
979 type: "user",
980 payload: { text: "<prompt>" },
981 delivery: "steer",
982 },
983 }),
984 event("session.execution.started", { sessionID: parent }),
985 event("session.text.ended", {
986 sessionID: parent,
987 assistantMessageID: "msg_next_reply",
988 ordinal: 0,
989 text: "NEXT",
990 }),
991 event("session.execution.succeeded", { sessionID: parent }),
992 ],
993 commands: (thread) => [
994 thread.create,
995 thread.message("start", bigPickle, BACKGROUND_PROMPT),
996 thread.message("next"),
997 ],
998 steps: (thread, run) => [
999 { type: "dispatch", command: thread.create },
1000 { type: "advance_clock", duration: "1 millis" },
1001 { type: "dispatch", command: thread.message("start", bigPickle, BACKGROUND_PROMPT) },
1002 { type: "advance_clock", duration: "1 millis" },
1003 { type: "await_thread_idle", threadId: thread.threadId },
1004 { type: "dispatch", command: thread.message("next") },
1005 { type: "advance_clock", duration: "1 millis" },
1006 {
1007 type: "await_run_status",
1008 threadId: thread.threadId,
1009 runId: run(2),
1010 status: "completed",
1011 },
1012 { type: "await_thread_idle", threadId: thread.threadId },
1013 ],
1014 });
1015 // The parent's run ended; the lost subagent ended as interrupted and
1016 // said why; no follow-up is waited on; the next turn ran normally.
1017 assert.deepEqual(
1018 projection.runs.map((candidate) => candidate.status),
1019 ["completed", "completed"],
1020 );
1021 assert.lengthOf(projection.subagents, 1);
1022 assert.deepInclude(projection.subagents[0], { status: "interrupted" });
1023 assert.include(projection.subagents[0]?.result ?? "", "lost its connection to OpenCode");
1024 assert.isFalse(
1025 projection.turnItems.some(
1026 (item) => item.status === "running" || item.status === "waiting",
1027 ),
1028 );
1029 const shell = result.shellSnapshot.threads.find((row) => row.id === projection.thread.id);
1030 assert.deepEqual(shell?.pendingBackgroundTasks ?? [], []);
1031 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
1032 );
1033
1034 it.effect(
1035 "settles a background subagent whose report OpenCode delivers into the launching turn",
1036 () =>
1037 Effect.gen(function* () {
1038 // The recorded background run, reordered as 2.0.18 ran it live when the
1039 // child ended first: the parent's execution is still answering when the
1040 // child's report is queued and delivered into it, and that one
1041 // execution then answers the report and ends. No execution starts on
1042 // its own, so no continuation run opens.
1043 const cwd = yield* checkpointWorkspace("opencode2-background-in-turn");
1044 const recorded = yield* readProviderReplayTranscript(
1045 new URL(
1046 "./testkit/fixtures/opencode2_background/opencode_transcript.ndjson",
1047 import.meta.url,
1048 ),
1049 );
1050 const byLabel = (label: string) => {
1051 const entry = recorded.entries.find(
1052 (candidate) => candidate.type !== "runtime_exit" && candidate.label === label,
1053 );
1054 assert.isDefined(entry);
1055 return entry!;
1056 };
1057 const parentEnd = recorded.entries.findIndex(
1058 (entry) => entry.type !== "runtime_exit" && entry.label === "session.execution.succeeded",
1059 );
1060 const childStart = recorded.entries.findIndex(
1061 (entry) => entry.type !== "runtime_exit" && entry.label === "session.step.started.3",
1062 );
1063 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
1064 ...recorded,
1065 scenario: "opencode2-background-in-turn",
1066 entries: [
1067 // The launch, up to the parent's answer, without its end.
1068 ...recorded.entries.slice(0, parentEnd),
1069 // The child runs to its end meanwhile.
1070 ...recorded.entries.slice(
1071 childStart,
1072 recorded.entries.indexOf(byLabel("session.inbox.enqueued.3")),
1073 ),
1074 // Its report is queued and delivered into the parent's running execution.
1075 byLabel("session.inbox.enqueued.3"),
1076 byLabel("session.inbox.delivered.3"),
1077 // That execution answers the report and ends.
1078 byLabel("session.text.ended.3"),
1079 byLabel("session.execution.succeeded.3"),
1080 ],
1081 });
1082 const thread = threadCommands({ name: "opencode2-background-in-turn", worktreePath: cwd });
1083 const ids = yield* IdAllocator.IdAllocatorV2;
1084 const result = yield* runOrchestratorV2ProviderReplayScenario(
1085 {
1086 name: "opencode2-background-in-turn",
1087 transcript,
1088 commands: [thread.create, thread.message("start", bigPickle, BACKGROUND_PROMPT)],
1089 steps: [
1090 { type: "dispatch", command: thread.create },
1091 { type: "advance_clock", duration: "1 millis" },
1092 { type: "dispatch", command: thread.message("start", bigPickle, BACKGROUND_PROMPT) },
1093 { type: "advance_clock", duration: "1 millis" },
1094 {
1095 type: "await_run_status",
1096 threadId: thread.threadId,
1097 runId: ids.derive.run({ threadId: thread.threadId, ordinal: 1 }),
1098 status: "completed",
1099 },
1100 { type: "await_thread_idle", threadId: thread.threadId },
1101 ],
1102 },
1103 OpenCode2OrchestratorReplayHarness,
1104 { runContinuationWorker: true },
1105 ).pipe(provideDeterministicTestRuntime);
1106 const projection = result.projections.get(thread.threadId);
1107 assert.isDefined(projection);
1108 // One run took the launch and the report; the subagent completed with
1109 // its output, and nothing waits on a follow-up.
1110 assert.deepEqual(
1111 projection!.runs.map((candidate) => candidate.status),
1112 ["completed"],
1113 );
1114 assert.lengthOf(projection!.subagents, 1);
1115 assert.deepInclude(projection!.subagents[0], { status: "completed" });
1116 assert.include(projection!.subagents[0]?.result ?? "", "CHILD_OK");
1117 assert.isTrue(
1118 projection!.turnItems.some(
1119 (item) =>
1120 item.type === "assistant_message" && item.text.includes("finished and returned"),
1121 ),
1122 );
1123 assert.isFalse(
1124 projection!.turnItems.some(
1125 (item) => item.status === "running" || item.status === "waiting",
1126 ),
1127 );
1128 const shell = result.shellSnapshot.threads.find((row) => row.id === projection!.thread.id);
1129 assert.deepEqual(shell?.pendingBackgroundTasks ?? [], []);
1130 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
1131 );
1132
1133 it.effect(
1134 "runs plan mode as OpenCode's plan agent and switches back before the next prompt",
1135 () =>
1136 Effect.gen(function* () {
1137 const name = "opencode2_switch";
1138 const cwd = yield* checkpointWorkspace(name);
1139 const thread = threadCommands({ name, worktreePath: cwd });
1140 // The spike's recording: plan agent, then build agent and a new model on
1141 // one session. Its `<work>` is this test's workspace.
1142 const recorded = yield* Effect.gen(function* () {
1143 const fs = yield* FileSystem.FileSystem;
1144 const path = yield* Path.Path;
1145 const text = yield* fs.readFileString(
1146 yield* path.fromFileUrl(
1147 new URL(
1148 "./testkit/fixtures/opencode2_switch/opencode_transcript.ndjson",
1149 import.meta.url,
1150 ),
1151 ),
1152 );
1153 return yield* decodeProviderReplayNdjson(text.replaceAll("<work>", cwd));
1154 }).pipe(Effect.provide(NodeServices.layer));
1155 const projection = yield* runScenario({
1156 name,
1157 threadId: thread.threadId,
1158 entries: recorded.entries,
1159 commands: [
1160 thread.create,
1161 thread.interactionMode("mode-plan", "plan"),
1162 thread.message(
1163 "plan",
1164 bigPickle,
1165 "Create a file named plan_probe.txt containing HI using the write tool.",
1166 ),
1167 thread.interactionMode("mode-default", "default"),
1168 thread.message("switched", nemotron, "Reply exactly SWITCHED."),
1169 ],
1170 });
1171 assert.deepEqual(
1172 projection.runs.map((run) => [run.status, run.modelSelection.model]),
1173 [
1174 ["completed", bigPickle.model],
1175 ["completed", nemotron.model],
1176 ],
1177 );
1178 const replies = projection.turnItems.flatMap((item) =>
1179 item.type === "assistant_message" ? [item.text] : [],
1180 );
1181 // The plan agent refused to write; OpenCode's own reminder told it why.
1182 assert.include(replies[0], "Plan mode");
1183 assert.equal(replies[1], "SWITCHED");
1184 assert.lengthOf(projection.providerThreads, 1);
1185 }).pipe(Effect.scoped),
1186 );
1187});