1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, describe, it } from "@effect/vitest";
3import {
4 CommandId,
5 MessageId,
6 type ModelSelection,
7 type OrchestrationV2Command,
8 type OrchestrationV2Run,
9 ProjectId,
10 type ProviderInteractionMode,
11 ProviderInstanceId,
12 type ProviderReplayEntry,
13 type RuntimeMode,
14 ThreadId,
15} from "@t3tools/contracts";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18
19import {
20 OPENCODE2_HTTP_PROTOCOL,
21 OpenCode2OrchestratorReplayHarness,
22} from "./Adapters/OpenCode2AdapterV2.testkit.ts";
23import { OPENCODE_PROVIDER } from "@t3tools/provider-opencode/testing";
24import { provideDeterministicTestRuntime } from "./testkit/DeterministicRuntime.ts";
25import type { OrchestratorV2ScenarioStep } from "./testkit/OrchestratorScenario.ts";
26import { runOrchestratorV2ProviderReplayScenario } from "./testkit/ProviderReplayHarness.ts";
27import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
28import { readProviderReplayTranscript } from "@t3tools/provider-testing/replayTranscript";
29import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
30
31const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX";
32
33const FIRST_TURN_END = "first-turn-end";
34
35const BACKGROUND_HOLD = "background-hold";
36
37const BACKGROUND_PARENT = "ses_f1485cda4ffeXuc6GjAU9vDiRb";
38
39const BACKGROUND_CHILD = "ses_f1485c529ffea4URrYruwEg0Ja";
40const BACKGROUND_PROMPT =
41 "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.";
42const instanceId = ProviderInstanceId.make("opencode");
43const bigPickle: ModelSelection = { instanceId, model: "opencode/big-pickle" };
44const mimo: ModelSelection = { instanceId, model: "opencode/mimo-v2.6-flash-free" };
45
46const out = (type: string, input?: unknown): ProviderReplayEntry => ({
47 type: "expect_outbound",
48 frame: input === undefined ? { type } : { type, input },
49});
50const reply = (operation: string, data: unknown): ProviderReplayEntry => ({
51 type: "emit_inbound",
52 frame: { type: "sdk.response", operation, data },
53});
54const event = (type: string, data: Record<string, unknown>): ProviderReplayEntry => ({
55 type: "emit_inbound",
56 frame: {
57 type: "sdk.event",
58 event: { id: `evt_${type.replaceAll(".", "")}`, created: 1, type, data },
59 },
60});
61const labelled = (entry: ProviderReplayEntry, label: string): ProviderReplayEntry =>
62 entry.type === "runtime_exit" ? entry : { ...entry, label };
63/** MCP rules are last-match-wins; allow this thread after denying all T3 servers. */
64const mcpRules = (name: string) => [
65 { action: "t3-code-*", resource: "*", effect: "deny" },
66 { action: `t3-code-thread_${name}_*`, resource: "*", effect: "allow" },
67];
68const FULL_ACCESS = [{ action: "*", resource: "*", effect: "allow" }];
69
70const t3Rules = (name: string) => [...FULL_ACCESS, ...mcpRules(name)];
71
72const BUILD_PATHS = [
73 {
74 action: "external_directory",
75 resource: "/home/.local/share/opencode/tool-output/*",
76 effect: "allow",
77 },
78];
79const PLAN_PATHS = [
80 ...BUILD_PATHS,
81 { action: "edit", resource: "/home/.opencode/plan/*", effect: "allow" },
82 { action: "external_directory", resource: "/home/.opencode/plan/*", effect: "allow" },
83];
84const agentInfo = (id: string, description: string, permissions: ReadonlyArray<unknown>) => ({
85 id,
86 name: id === "plan" ? "Plan" : "Build",
87 request: { settings: {}, headers: {}, body: {} },
88 description,
89 mode: "primary",
90 hidden: false,
91 permissions,
92});
93
94const agentList = (directory: string) => ({
95 location: { directory },
96 data: [
97 agentInfo("build", "The default agent.", [
98 { action: "*", resource: "*", effect: "allow" },
99 { action: "external_directory", resource: "*", effect: "ask" },
100 ...BUILD_PATHS,
101 ]),
102 agentInfo("plan", "Read-only agent for planning.", [
103 { action: "*", resource: "*", effect: "allow" },
104 { action: "external_directory", resource: "*", effect: "ask" },
105 ...BUILD_PATHS,
106 { action: "edit", resource: "*", effect: "deny" },
107 ...PLAN_PATHS.slice(BUILD_PATHS.length),
108 ]),
109 ],
110});
111
112const autoEditRules = (name: string) => [
113 { action: "shell", resource: "*", effect: "ask" },
114 { action: "edit", resource: "*", effect: "allow" },
115 { action: "external_directory", resource: "*", effect: "ask" },
116 ...BUILD_PATHS,
117 ...mcpRules(name),
118];
119
120const planRules = (name: string) => [
121 { action: "*", resource: "*", effect: "allow" },
122 { action: "edit", resource: "*", effect: "deny" },
123 ...PLAN_PATHS,
124 ...mcpRules(name),
125];
126
127const sessionInfo = (directory: string, permissions: ReadonlyArray<unknown>, title: string) => ({
128 data: {
129 id: SESSION,
130 title,
131 projectID: "global",
132 model: { id: "big-pickle", providerID: "opencode", variant: "default" },
133 cost: 0,
134 tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
135 time: { created: 1790656601394, updated: 1790656601394 },
136 location: { directory },
137 permissions,
138 },
139});
140
141const instructionsWritten: ReadonlyArray<ProviderReplayEntry> = [
142 out("session.instructions.entry.put", { sessionID: SESSION, key: "t3-code", value: "<any>" }),
143 reply("session.instructions.entry.put", null),
144];
145
146const answeredPrompt = (text: string): ReadonlyArray<ProviderReplayEntry> => [
147 out("session.prompt", { sessionID: SESSION, text: "<any>" }),
148 reply("session.prompt", {
149 data: {
150 id: `msg_user_${text}`,
151 sessionID: SESSION,
152 time: { created: 1790656601410 },
153 type: "user",
154 payload: { text: "<prompt>" },
155 delivery: "steer",
156 },
157 }),
158 event("session.text.ended", {
159 sessionID: SESSION,
160 assistantMessageID: `msg_assistant_${text}`,
161 ordinal: 0,
162 text,
163 }),
164 event("session.execution.succeeded", { sessionID: SESSION }),
165];
166
167/** A `/api/model` entry as 2.0.18 lists it; a known window means no re-read after a turn. */
168const catalogModel = (id: string, name: string) => ({
169 id,
170 modelID: id,
171 providerID: "opencode",
172 family: id,
173 name,
174 compatibility: { reasoningField: "reasoning_content" },
175 package: "@opencode/ai/providers/openai-compatible",
176 settings: { apiKey: "public", baseURL: "https://opencode.ai/zen/v1", provider: "opencode" },
177 capabilities: { tools: true, input: ["text"], output: ["text"] },
178 variants: [],
179 time: { released: 1760659200000 },
180 cost: [{ input: 0, output: 0, cache: { read: 0, write: 0 } }],
181 status: "active",
182 enabled: true,
183 limit: { context: 200000, input: 160000, output: 32000 },
184});
185const createdSession = (
186 directory: string,
187 name: string,
188 permissions: ReadonlyArray<unknown> = t3Rules(name),
189
190 narrows = false,
191): ReadonlyArray<ProviderReplayEntry> => [
192 out("event.subscribe"),
193 out("model.list", "<any>"),
194 reply("model.list", {
195 location: { directory },
196 data: [
197 catalogModel("big-pickle", "Big Pickle"),
198 catalogModel("mimo-v2.6-flash-free", "MiMo V2.6 Flash Free"),
199 ],
200 }),
201 ...(narrows ? [out("agent.list", "<any>"), reply("agent.list", agentList(directory))] : []),
202
203 out("session.create", { location: { directory }, model: "<any>", permissions, title: name }),
204 reply("session.create", sessionInfo(directory, permissions, name)),
205];
206
207const withDirectory = <T>(value: T, directory: string): T => {
208 const replace = (entry: unknown): unknown =>
209 entry === "<work>"
210 ? directory
211 : Array.isArray(entry)
212 ? entry.map(replace)
213 : typeof entry === "object" && entry !== null
214 ? Object.fromEntries(Object.entries(entry).map(([key, inner]) => [key, replace(inner)]))
215 : entry;
216 return replace(value) as T;
217};
218
219const threadCommands = (input: {
220 readonly name: string;
221 readonly worktreePath: string;
222 readonly runtimeMode?: RuntimeMode;
223 readonly interactionMode?: ProviderInteractionMode;
224}) => {
225 const threadId = ThreadId.make(`thread:${input.name}`);
226 const command = (key: string) => CommandId.make(`command:${input.name}:${key}`);
227 return {
228 threadId,
229 create: {
230 type: "thread.create",
231 createdBy: "user",
232 creationSource: "web",
233 commandId: command("create"),
234 threadId,
235 projectId: ProjectId.make(`project:${input.name}`),
236 title: input.name,
237 modelSelection: bigPickle,
238 runtimeMode: input.runtimeMode ?? "full-access",
239 interactionMode: input.interactionMode ?? "default",
240 branch: null,
241 worktreePath: input.worktreePath,
242 } satisfies OrchestrationV2Command,
243 message: (key: string, modelSelection: ModelSelection = bigPickle, text?: string) =>
244 ({
245 type: "message.dispatch",
246 createdBy: "user",
247 creationSource: "web",
248 commandId: command(key),
249 threadId,
250 messageId: MessageId.make(`message:${input.name}:${key}`),
251 text: text ?? `Reply with exactly: ${key}`,
252 attachments: [],
253 modelSelection,
254 dispatchMode: { type: "start_immediately" },
255 }) satisfies OrchestrationV2Command,
256 interactionMode: (key: string, interactionMode: "default" | "plan") =>
257 ({
258 type: "thread.interaction-mode.set",
259 commandId: command(key),
260 threadId,
261 interactionMode,
262 }) satisfies OrchestrationV2Command,
263 command,
264 };
265};
266
267const runScenario = (input: {
268 readonly name: string;
269 readonly threadId: ThreadId;
270 readonly entries: ReadonlyArray<ProviderReplayEntry>;
271 readonly commands: ReadonlyArray<OrchestrationV2Command>;
272}) =>
273 Effect.gen(function* () {
274 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
275 provider: OPENCODE_PROVIDER,
276 protocol: OPENCODE2_HTTP_PROTOCOL,
277 version: "2.0.18",
278 scenario: input.name,
279 entries: input.entries,
280 });
281 const steps = input.commands.flatMap((command): Array<OrchestratorV2ScenarioStep> => [
282 { type: "dispatch", command },
283 { type: "advance_clock", duration: "1 millis" },
284 ...(command.type === "message.dispatch"
285 ? [{ type: "await_thread_idle" as const, threadId: input.threadId }]
286 : []),
287 ]);
288 const result = yield* runOrchestratorV2ProviderReplayScenario(
289 { name: input.name, transcript, commands: input.commands, steps },
290 OpenCode2OrchestratorReplayHarness,
291 ).pipe(provideDeterministicTestRuntime);
292 const projection = result.projections.get(input.threadId);
293 assert.isDefined(projection);
294 return projection;
295 });
296
297describe("OpenCode 2 through the orchestrator", () => {
298 it.effect.each(["message", "thread settings"] as const)(
299 "switches the session's model before the next prompt when changed from the %s",
300 (via) =>
301 Effect.gen(function* () {
302 const name = `opencode2-model-switch-${via.replace(" ", "-")}`;
303 const cwd = yield* checkpointWorkspace(name);
304 const thread = threadCommands({ name, worktreePath: cwd });
305 const projection = yield* runScenario({
306 name,
307 threadId: thread.threadId,
308 entries: [
309 ...createdSession(cwd, name),
310 ...instructionsWritten,
311 ...answeredPrompt("FIRST"),
312
313 out("session.get", { sessionID: SESSION }),
314 reply("session.get", sessionInfo(cwd, t3Rules(name), name)),
315 out("session.switchModel", {
316 sessionID: SESSION,
317 model: { providerID: "opencode", id: "mimo-v2.6-flash-free" },
318 }),
319 reply("session.switchModel", null),
320
321 ...instructionsWritten,
322 ...answeredPrompt("SECOND"),
323 ],
324 commands: [
325 thread.create,
326 thread.message("first"),
327 ...(via === "thread settings"
328 ? [
329 {
330 type: "thread.model-selection.set",
331 commandId: thread.command("model"),
332 threadId: thread.threadId,
333 modelSelection: mimo,
334 } satisfies OrchestrationV2Command,
335 ]
336 : []),
337 thread.message("second", mimo),
338 ],
339 });
340 assert.deepEqual(
341 projection.runs.map((run) => [run.status, run.modelSelection.model]),
342 [
343 ["completed", bigPickle.model],
344 ["completed", mimo.model],
345 ],
346 );
347
348 assert.lengthOf(projection.providerThreads, 1);
349 }),
350 );
351
352 it.effect("rewrites the session's rules when the thread's mode changes between turns", () =>
353 Effect.gen(function* () {
354 const name = "opencode2-mode-change";
355 const cwd = yield* checkpointWorkspace(name);
356 const thread = threadCommands({ name, worktreePath: cwd });
357 const setMode = (key: string, runtimeMode: RuntimeMode) =>
358 ({
359 type: "thread.runtime-mode.set",
360 commandId: thread.command(key),
361 threadId: thread.threadId,
362 runtimeMode,
363 }) satisfies OrchestrationV2Command;
364 const projection = yield* runScenario({
365 name,
366 threadId: thread.threadId,
367 entries: [
368 ...createdSession(cwd, name),
369 ...instructionsWritten,
370 ...answeredPrompt("FIRST"),
371
372 out("session.get", { sessionID: SESSION }),
373 reply("session.get", sessionInfo(cwd, t3Rules(name), name)),
374 out("agent.list", "<any>"),
375 reply("agent.list", agentList(cwd)),
376 out("session.update", { sessionID: SESSION, permissions: autoEditRules(name) }),
377 reply("session.update", null),
378 ...answeredPrompt("SECOND"),
379
380 out("session.get", { sessionID: SESSION }),
381 reply("session.get", sessionInfo(cwd, autoEditRules(name), name)),
382 out("session.update", { sessionID: SESSION, permissions: t3Rules(name) }),
383 reply("session.update", null),
384 ...answeredPrompt("THIRD"),
385 ],
386 commands: [
387 thread.create,
388 thread.message("first"),
389 setMode("auto-edit", "auto-accept-edits"),
390 thread.message("second"),
391 setMode("full", "full-access"),
392 thread.message("third"),
393 ],
394 });
395 assert.deepEqual(
396 projection.runs.map((run) => run.status),
397 ["completed", "completed", "completed"],
398 );
399 assert.lengthOf(projection.providerThreads, 1);
400 }),
401 );
402
403 it.effect("denies edits outside the plan directory in plan mode and lifts it after", () =>
404 Effect.gen(function* () {
405 const name = "opencode2-plan-rules";
406 const cwd = yield* checkpointWorkspace(name);
407 const thread = threadCommands({ name, worktreePath: cwd, interactionMode: "plan" });
408 const projection = yield* runScenario({
409 name,
410 threadId: thread.threadId,
411 entries: [
412 ...createdSession(cwd, name, planRules(name), true),
413
414 out("session.switchAgent", { sessionID: SESSION, agent: "plan" }),
415 reply("session.switchAgent", null),
416 ...instructionsWritten,
417 ...answeredPrompt("PLANNED"),
418 out("session.get", { sessionID: SESSION }),
419 reply("session.get", sessionInfo(cwd, planRules(name), name)),
420 out("session.update", { sessionID: SESSION, permissions: t3Rules(name) }),
421 reply("session.update", null),
422 out("session.switchAgent", { sessionID: SESSION, agent: "build" }),
423 reply("session.switchAgent", null),
424 ...answeredPrompt("BUILT"),
425 ],
426 commands: [
427 thread.create,
428 thread.message("plan"),
429 {
430 type: "thread.interaction-mode.set",
431 commandId: thread.command("default"),
432 threadId: thread.threadId,
433 interactionMode: "default",
434 },
435 thread.message("build"),
436 ],
437 });
438 assert.deepEqual(
439 projection.runs.map((run) => run.status),
440 ["completed", "completed"],
441 );
442 }),
443 );
444
445 it.effect("forks from an earlier turn before the next turn's user message", () =>
446 Effect.gen(function* () {
447 const name = "opencode2-fork";
448 const cwd = yield* checkpointWorkspace(name);
449 const recorded = yield* readProviderReplayTranscript(
450 new URL("./testkit/fixtures/opencode2_fork/opencode_transcript.ndjson", import.meta.url),
451 );
452
453 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript(
454 withDirectory(recorded, cwd),
455 );
456 const source = threadCommands({ name, worktreePath: cwd });
457 const target = ThreadId.make(`thread:${name}:target`);
458 const [one, two] = [source.message("one"), source.message("two")];
459 const commands: ReadonlyArray<OrchestrationV2Command> = [
460 source.create,
461 one,
462 two,
463 {
464 type: "thread.fork",
465 createdBy: "user",
466 creationSource: "web",
467 commandId: source.command("fork"),
468 sourceThreadId: source.threadId,
469 targetThreadId: target,
470 sourcePoint: {
471 type: "run",
472 runId: (yield* IdAllocator.IdAllocatorV2).derive.run({
473 threadId: source.threadId,
474 ordinal: 1,
475 }),
476 },
477 },
478 {
479 ...source.message("repeat"),
480 threadId: target,
481 messageId: MessageId.make(`message:${name}:repeat`),
482 },
483 ];
484 const steps: Array<OrchestratorV2ScenarioStep> = commands.flatMap((command) => [
485 { type: "dispatch" as const, command },
486 { type: "advance_clock" as const, duration: "1 millis" as const },
487 ...(command.type === "message.dispatch"
488 ? [{ type: "await_thread_idle" as const, threadId: command.threadId }]
489 : []),
490 ]);
491 const result = yield* runOrchestratorV2ProviderReplayScenario(
492 { name, transcript, commands, steps, projectionThreadIds: [source.threadId, target] },
493 OpenCode2OrchestratorReplayHarness,
494 ).pipe(provideDeterministicTestRuntime);
495 const forked = result.projections.get(target);
496 assert.isDefined(forked);
497 assert.equal(forked.contextTransfers[0]?.resolution?.strategy, "native_fork");
498 assert.equal(
499 forked.providerThreads[0]?.nativeThreadRef?.nativeId,
500 recorded.metadata?.["forkedNativeSessionId"],
501 );
502
503 assert.deepEqual(
504 forked.runs.map((run) => run.status),
505 ["completed"],
506 );
507 const visible = forked.visibleTurnItems.flatMap((row) =>
508 row.item.type === "assistant_message" ? [row.item.text] : [],
509 );
510 assert.deepEqual(visible, ["ONE", "ONE"]);
511 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
512 );
513
514 it.effect("never sends OpenCode a queued message that was cancelled", () =>
515 Effect.gen(function* () {
516 const name = "opencode2-queued-cancel";
517 const cwd = yield* checkpointWorkspace(name);
518 const thread = threadCommands({ name, worktreePath: cwd });
519 const queued = (key: string): OrchestrationV2Command => ({
520 ...thread.message(key),
521 dispatchMode: { type: "queue_after_active" },
522 });
523 const ids = yield* IdAllocator.IdAllocatorV2;
524 const run = (ordinal: number) => ids.derive.run({ threadId: thread.threadId, ordinal });
525
526 const [first, second, third] = [thread.message("first"), queued("second"), queued("third")];
527 const cancel: OrchestrationV2Command = {
528 type: "queued-run.cancel",
529 commandId: thread.command("cancel"),
530 threadId: thread.threadId,
531 runId: run(2),
532 };
533 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
534 provider: OPENCODE_PROVIDER,
535 protocol: OPENCODE2_HTTP_PROTOCOL,
536 version: "2.0.18",
537 scenario: name,
538 entries: [
539 ...createdSession(cwd, name),
540 ...instructionsWritten,
541 out("session.prompt", { sessionID: SESSION, id: "<any>", text: "<any>" }),
542 reply("session.prompt", {
543 data: {
544 id: "msg_user_FIRST",
545 sessionID: SESSION,
546 time: { created: 1 },
547 type: "user",
548 payload: { text: "<prompt>" },
549 delivery: "steer",
550 },
551 }),
552
553 ...answeredPrompt("THIRD"),
554 ],
555 });
556 const steps: Array<OrchestratorV2ScenarioStep> = [
557 { type: "dispatch", command: thread.create },
558 { type: "advance_clock", duration: "1 millis" },
559 { type: "dispatch", command: first, await: false, key: "first" },
560 { type: "await_run_steerable", threadId: thread.threadId, runId: run(1) },
561 { type: "dispatch", command: second },
562 { type: "dispatch", command: third },
563 { type: "dispatch", command: cancel },
564 { type: "await_run_status", threadId: thread.threadId, runId: run(2), status: "cancelled" },
565 { type: "release_replay_gate", label: FIRST_TURN_END },
566 { type: "await", key: "first" },
567 { type: "await_thread_idle", threadId: thread.threadId },
568 ];
569 const result = yield* runOrchestratorV2ProviderReplayScenario(
570 {
571 name,
572 transcript: {
573 ...transcript,
574 entries: [
575 ...transcript.entries.slice(0, -4),
576 labelled(
577 event("session.execution.succeeded", { sessionID: SESSION }),
578 FIRST_TURN_END,
579 ),
580 ...transcript.entries.slice(-4),
581 ],
582 },
583 commands: [thread.create, first, second, third, cancel],
584 steps,
585 },
586 OpenCode2OrchestratorReplayHarness,
587 ).pipe(provideDeterministicTestRuntime);
588 const projection = result.projections.get(thread.threadId);
589 assert.isDefined(projection);
590 assert.deepEqual(
591 projection.runs.map((candidate) => candidate.status),
592 ["completed", "cancelled", "completed"],
593 );
594 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
595 );
596
597 const backgroundReconnect = Effect.fn("backgroundReconnect")(function* (input: {
598 readonly name: string;
599 readonly cut: string;
600 readonly hold?: string;
601 readonly reconnect: (parent: string) => ReadonlyArray<ProviderReplayEntry>;
602 readonly steps: (
603 thread: ReturnType<typeof threadCommands>,
604 run: (ordinal: number) => OrchestrationV2Run["id"],
605 ) => ReadonlyArray<OrchestratorV2ScenarioStep>;
606 readonly commands: (
607 thread: ReturnType<typeof threadCommands>,
608 ) => ReadonlyArray<OrchestrationV2Command>;
609 }) {
610 const cwd = yield* checkpointWorkspace(input.name);
611 const recorded = yield* readProviderReplayTranscript(
612 new URL(
613 "./testkit/fixtures/opencode2_background/opencode_transcript.ndjson",
614 import.meta.url,
615 ),
616 );
617 const cut = recorded.entries.findIndex(
618 (entry) => entry.type !== "runtime_exit" && entry.label === input.cut,
619 );
620 assert.isAtLeast(cut, 0);
621 const kept = recorded.entries
622 .slice(0, cut)
623 .map((entry) =>
624 entry.type !== "runtime_exit" && entry.label === input.hold
625 ? labelled(entry, BACKGROUND_HOLD)
626 : entry,
627 );
628 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
629 ...recorded,
630 scenario: input.name,
631 entries: [
632 ...kept,
633 { type: "runtime_exit", status: "success" },
634 ...input.reconnect(BACKGROUND_PARENT),
635 ],
636 });
637 const thread = threadCommands({ name: input.name, worktreePath: cwd });
638 const ids = yield* IdAllocator.IdAllocatorV2;
639 const run = (ordinal: number) => ids.derive.run({ threadId: thread.threadId, ordinal });
640 const result = yield* runOrchestratorV2ProviderReplayScenario(
641 {
642 name: input.name,
643 transcript,
644 commands: input.commands(thread),
645 steps: input.steps(thread, run),
646 },
647 OpenCode2OrchestratorReplayHarness,
648
649 { runContinuationWorker: true },
650 ).pipe(provideDeterministicTestRuntime);
651 const projection = result.projections.get(thread.threadId);
652 assert.isDefined(projection);
653 return { result, projection };
654 });
655
656 it.effect("backfills a follow-up whose stream dropped mid-run and ends it completed", () =>
657 Effect.gen(function* () {
658 const { result, projection } = yield* backgroundReconnect({
659 name: "opencode2-follow-up-reconnect",
660
661 cut: "session.step.streamed.5",
662 hold: "session.text.ended.3",
663 reconnect: (parent) => [
664 out("event.subscribe"),
665 out("session.active"),
666 reply("session.active", { data: {} }),
667
668 out("message.list", { sessionID: parent, order: "desc", limit: "50" }),
669 reply("message.list", {
670 data: [
671 {
672 id: "msg_0eb7a99c0001idleFollowUp00",
673 time: { created: 5 },
674 type: "idle",
675 outcome: "succeeded",
676 },
677 {
678 id: "msg_0eb7a96bb00170K1f1HJUeBNwF",
679 time: { created: 4 },
680 type: "assistant",
681 agent: "build",
682 model: { id: "big-pickle", providerID: "opencode", variant: "default" },
683 content: [
684 {
685 type: "text",
686 text: "The background subagent finished and returned `CHILD_OK`.",
687 },
688 ],
689 finish: "stop",
690 },
691 {
692 id: "msg_0eb7a3aee0015KmTB8XkMDW8iy",
693 time: { created: 3 },
694 type: "synthetic",
695 text: "<subagent>CHILD_OK</subagent>",
696 },
697 {
698 id: "msg_0eb7a3e3c001idleParentTurn0",
699 time: { created: 2 },
700 type: "idle",
701 outcome: "succeeded",
702 },
703 ],
704 cursor: {},
705 }),
706 ],
707 commands: (thread) => [
708 thread.create,
709 thread.message("start", bigPickle, BACKGROUND_PROMPT),
710 ],
711 steps: (thread, run) => [
712 { type: "dispatch", command: thread.create },
713 { type: "advance_clock", duration: "1 millis" },
714
715 {
716 type: "dispatch",
717 command: thread.message("start", bigPickle, BACKGROUND_PROMPT),
718 await: false,
719 key: "start",
720 },
721 { type: "advance_clock", duration: "1 millis" },
722
723 { type: "await_run_steerable", threadId: thread.threadId, runId: run(2) },
724 { type: "release_replay_gate", label: BACKGROUND_HOLD },
725 {
726 type: "await_run_status",
727 threadId: thread.threadId,
728 runId: run(2),
729 status: "completed",
730 },
731 { type: "await", key: "start" },
732 { type: "await_thread_idle", threadId: thread.threadId },
733 ],
734 });
735 assert.deepEqual(
736 projection.runs.map((candidate) => candidate.status),
737 ["completed", "completed"],
738 );
739 const followUp = projection.runs[1];
740 assert.deepEqual(
741 projection.turnItems.flatMap((item) =>
742 item.runId === followUp?.id && item.type === "assistant_message" ? [item.text] : [],
743 ),
744 ["The background subagent finished and returned `CHILD_OK`."],
745 );
746 const shell = result.shellSnapshot.threads.find((row) => row.id === projection.thread.id);
747 assert.deepEqual(shell?.pendingBackgroundTasks ?? [], []);
748 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
749 );
750
751 it.effect(
752 "settles a background subagent whose end was lost with the stream, and takes the next turn",
753 () =>
754 Effect.gen(function* () {
755 const { result, projection } = yield* backgroundReconnect({
756 name: "opencode2-background-lost",
757
758 cut: "session.step.started.3",
759 reconnect: (parent) => [
760 out("event.subscribe"),
761
762 out("session.active"),
763 reply("session.active", { data: {} }),
764
765 out("message.list", { sessionID: BACKGROUND_CHILD, order: "desc", limit: "50" }),
766 reply("message.list", {
767 data: [
768 {
769 id: "msg_0eb7a99b0001idleChildTurn0",
770 time: { created: 4 },
771 type: "idle",
772 outcome: "succeeded",
773 },
774 {
775 id: "msg_0eb7a916c001NRJQJ5gwIT0QnU",
776 time: { created: 3 },
777 type: "assistant",
778 agent: "general",
779 model: { id: "big-pickle", providerID: "opencode", variant: "default" },
780 content: [{ type: "text", text: "CHILD_OK" }],
781 finish: "stop",
782 },
783 {
784 id: "msg_0eb7a3aea001Lw1Pav5d4akfEg",
785 time: { created: 2 },
786 type: "user",
787 text: "Run the shell command `sleep 20` with the bash tool and then reply exactly CHILD_OK.",
788 },
789 ],
790 cursor: {},
791 }),
792 out("session.prompt", { sessionID: parent, id: "<any>", text: "<any>" }),
793 reply("session.prompt", {
794 data: {
795 id: "msg_next",
796 sessionID: parent,
797 time: { created: 9 },
798 type: "user",
799 payload: { text: "<prompt>" },
800 delivery: "steer",
801 },
802 }),
803 event("session.execution.started", { sessionID: parent }),
804 event("session.text.ended", {
805 sessionID: parent,
806 assistantMessageID: "msg_next_reply",
807 ordinal: 0,
808 text: "NEXT",
809 }),
810 event("session.execution.succeeded", { sessionID: parent }),
811 ],
812 commands: (thread) => [
813 thread.create,
814 thread.message("start", bigPickle, BACKGROUND_PROMPT),
815 thread.message("next"),
816 ],
817 steps: (thread, run) => [
818 { type: "dispatch", command: thread.create },
819 { type: "advance_clock", duration: "1 millis" },
820 { type: "dispatch", command: thread.message("start", bigPickle, BACKGROUND_PROMPT) },
821 { type: "advance_clock", duration: "1 millis" },
822 { type: "await_thread_idle", threadId: thread.threadId },
823 { type: "dispatch", command: thread.message("next") },
824 { type: "advance_clock", duration: "1 millis" },
825 {
826 type: "await_run_status",
827 threadId: thread.threadId,
828 runId: run(2),
829 status: "completed",
830 },
831 { type: "await_thread_idle", threadId: thread.threadId },
832 ],
833 });
834
835 assert.deepEqual(
836 projection.runs.map((candidate) => candidate.status),
837 ["completed", "completed"],
838 );
839 assert.lengthOf(projection.subagents, 1);
840 assert.deepInclude(projection.subagents[0], { status: "interrupted" });
841 assert.include(projection.subagents[0]?.result ?? "", "lost its connection to OpenCode");
842 assert.isFalse(
843 projection.turnItems.some(
844 (item) => item.status === "running" || item.status === "waiting",
845 ),
846 );
847 const shell = result.shellSnapshot.threads.find((row) => row.id === projection.thread.id);
848 assert.deepEqual(shell?.pendingBackgroundTasks ?? [], []);
849 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
850 );
851
852 it.effect(
853 "settles a background subagent whose report OpenCode delivers into the launching turn",
854 () =>
855 Effect.gen(function* () {
856 // OpenCode 2.0.18 can deliver a child report into the parent's still-running execution, which then answers it.
857
858 const cwd = yield* checkpointWorkspace("opencode2-background-in-turn");
859 const recorded = yield* readProviderReplayTranscript(
860 new URL(
861 "./testkit/fixtures/opencode2_background/opencode_transcript.ndjson",
862 import.meta.url,
863 ),
864 );
865 const byLabel = (label: string) => {
866 const entry = recorded.entries.find(
867 (candidate) => candidate.type !== "runtime_exit" && candidate.label === label,
868 );
869 assert.isDefined(entry);
870 return entry!;
871 };
872 const parentEnd = recorded.entries.findIndex(
873 (entry) => entry.type !== "runtime_exit" && entry.label === "session.execution.succeeded",
874 );
875 const childStart = recorded.entries.findIndex(
876 (entry) => entry.type !== "runtime_exit" && entry.label === "session.step.started.3",
877 );
878 const transcript = yield* OpenCode2OrchestratorReplayHarness.decodeTranscript({
879 ...recorded,
880 scenario: "opencode2-background-in-turn",
881 entries: [
882 ...recorded.entries.slice(0, parentEnd),
883
884 ...recorded.entries.slice(
885 childStart,
886 recorded.entries.indexOf(byLabel("session.inbox.enqueued.3")),
887 ),
888
889 byLabel("session.inbox.enqueued.3"),
890 byLabel("session.inbox.delivered.3"),
891
892 byLabel("session.text.ended.3"),
893 byLabel("session.execution.succeeded.3"),
894 ],
895 });
896 const thread = threadCommands({ name: "opencode2-background-in-turn", worktreePath: cwd });
897 const ids = yield* IdAllocator.IdAllocatorV2;
898 const result = yield* runOrchestratorV2ProviderReplayScenario(
899 {
900 name: "opencode2-background-in-turn",
901 transcript,
902 commands: [thread.create, thread.message("start", bigPickle, BACKGROUND_PROMPT)],
903 steps: [
904 { type: "dispatch", command: thread.create },
905 { type: "advance_clock", duration: "1 millis" },
906 { type: "dispatch", command: thread.message("start", bigPickle, BACKGROUND_PROMPT) },
907 { type: "advance_clock", duration: "1 millis" },
908 {
909 type: "await_run_status",
910 threadId: thread.threadId,
911 runId: ids.derive.run({ threadId: thread.threadId, ordinal: 1 }),
912 status: "completed",
913 },
914 { type: "await_thread_idle", threadId: thread.threadId },
915 ],
916 },
917 OpenCode2OrchestratorReplayHarness,
918 { runContinuationWorker: true },
919 ).pipe(provideDeterministicTestRuntime);
920 const projection = result.projections.get(thread.threadId);
921 assert.isDefined(projection);
922
923 assert.deepEqual(
924 projection!.runs.map((candidate) => candidate.status),
925 ["completed"],
926 );
927 assert.lengthOf(projection!.subagents, 1);
928 assert.deepInclude(projection!.subagents[0], { status: "completed" });
929 assert.include(projection!.subagents[0]?.result ?? "", "CHILD_OK");
930 assert.isTrue(
931 projection!.turnItems.some(
932 (item) =>
933 item.type === "assistant_message" && item.text.includes("finished and returned"),
934 ),
935 );
936 assert.isFalse(
937 projection!.turnItems.some(
938 (item) => item.status === "running" || item.status === "waiting",
939 ),
940 );
941 const shell = result.shellSnapshot.threads.find((row) => row.id === projection!.thread.id);
942 assert.deepEqual(shell?.pendingBackgroundTasks ?? [], []);
943 }).pipe(Effect.scoped, Effect.provide(Layer.merge(NodeServices.layer, IdAllocator.layer))),
944 );
945});