| ... | ... | @@ -953,6 +953,199 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { |
| 953 | 953 | }).pipe(Effect.scoped), |
| 954 | 954 | ); |
| 955 | 955 | |
| 956 | /** A `subagent` call's result while its subagent runs on (`session.tool.success` is at version 2). */ |
| 957 | const returnedRunning = (id: string): ProviderReplayEntry => ({ |
| 958 | type: "emit_inbound", |
| 959 | frame: { |
| 960 | type: "sdk.event", |
| 961 | event: { |
| 962 | id: `evt_toolsuccess_${id}`, |
| 963 | created: 1, |
| 964 | type: "session.tool.success", |
| 965 | data: { |
| 966 | sessionID: SESSION, |
| 967 | assistantMessageID: "msg_assistant", |
| 968 | id, |
| 969 | content: [{ type: "text", text: "The subagent is working in the background." }], |
| 970 | metadata: { sessionID: CHILD, status: "running" }, |
| 971 | executed: true, |
| 972 | }, |
| 973 | durable: { aggregateID: SESSION, seq: 1, version: 2 }, |
| 974 | }, |
| 975 | }, |
| 976 | }); |
| 977 | /** The model calls the subagent again by its session; the tool returns while it runs on. */ |
| 978 | const callAgain = (input: Record<string, unknown>): ReadonlyArray<ProviderReplayEntry> => { |
| 979 | const tool = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-again" }; |
| 980 | return [ |
| 981 | event("session.tool.input.started", { ...tool, name: "subagent" }), |
| 982 | event("session.tool.called", { |
| 983 | ...tool, |
| 984 | name: "subagent", |
| 985 | input: { description: "Again", prompt: "also this", sessionID: CHILD, ...input }, |
| 986 | executed: false, |
| 987 | }), |
| 988 | event("session.tool.progress", { |
| 989 | ...tool, |
| 990 | metadata: { sessionID: CHILD, status: "running" }, |
| 991 | }), |
| 992 | // A session called again gets the thread's rules. |
| 993 | out("session.update", { sessionID: CHILD, permissions: "<any>" }), |
| 994 | reply("session.update", null), |
| 995 | returnedRunning(tool.id), |
| 996 | ]; |
| 997 | }; |
| 998 | const childReport = (inboxID: string, text: string) => |
| 999 | event("session.inbox.enqueued", { |
| 1000 | inboxID, |
| 1001 | sessionID: SESSION, |
| 1002 | item: { |
| 1003 | type: "synthetic", |
| 1004 | payload: { |
| 1005 | text: `<subagent sessionID="${CHILD}" state="completed" description="Sleep">\n${text}\n</subagent>`, |
| 1006 | description: "Sleep", |
| 1007 | metadata: { source: "subagent", childID: CHILD, agent: "General", state: "completed" }, |
| 1008 | }, |
| 1009 | delivery: "steer", |
| 1010 | }, |
| 1011 | }); |
| 1012 | /** The subagent rows by title, the thread's background roster and the child's turns, as emitted. */ |
| 1013 | const watchCalls = ( |
| 1014 | runtime: ProviderAdapterV2SessionRuntime, |
| 1015 | thread: OrchestrationV2ProviderThread, |
| 1016 | ) => |
| 1017 | Effect.gen(function* () { |
| 1018 | const calls = new Map<string, { status: string; result: string | null }>(); |
| 1019 | const childTurns = new Map<string, string>(); |
| 1020 | const callsAtChildTurnStart = new Map< |
| 1021 | string, |
| 1022 | Record<string, { status: string; result: string | null }> |
| 1023 | >(); |
| 1024 | const state: { roster: ReadonlyArray<unknown> | undefined } = { roster: undefined }; |
| 1025 | let wake = yield* Deferred.make<void>(); |
| 1026 | yield* runtime.events.pipe( |
| 1027 | Stream.tap((event) => |
| 1028 | Effect.gen(function* () { |
| 1029 | if (event.type === "subagent.updated" && event.subagent.title !== null) { |
| 1030 | calls.set(event.subagent.title, { |
| 1031 | status: event.subagent.status, |
| 1032 | result: event.subagent.result, |
| 1033 | }); |
| 1034 | } |
| 1035 | if (event.type === "provider_thread.updated" && event.providerThread.id === thread.id) { |
| 1036 | state.roster = event.providerThread.pendingBackgroundTasks; |
| 1037 | } |
| 1038 | const id = |
| 1039 | event.type === "provider_turn.updated" |
| 1040 | ? event.providerTurn.nativeTurnRef?.nativeId |
| 1041 | : undefined; |
| 1042 | if ( |
| 1043 | event.type === "provider_turn.updated" && |
| 1044 | id?.startsWith(`${CHILD}:turn:`) === true |
| 1045 | ) { |
| 1046 | if (!childTurns.has(id)) { |
| 1047 | callsAtChildTurnStart.set(id, Object.fromEntries(calls)); |
| 1048 | } |
| 1049 | childTurns.set(id, event.providerTurn.status); |
| 1050 | } |
| 1051 | yield* Deferred.succeed(wake, undefined); |
| 1052 | }), |
| 1053 | ), |
| 1054 | Stream.runDrain, |
| 1055 | Effect.forkScoped, |
| 1056 | ); |
| 1057 | const until = (check: () => boolean): Effect.Effect<void> => |
| 1058 | Effect.suspend(() => { |
| 1059 | if (check()) return Effect.void; |
| 1060 | return Effect.gen(function* () { |
| 1061 | wake = yield* Deferred.make<void>(); |
| 1062 | if (check()) return; |
| 1063 | yield* Deferred.await(wake); |
| 1064 | yield* until(check); |
| 1065 | }); |
| 1066 | }); |
| 1067 | return { calls, callsAtChildTurnStart, childTurns, state, until }; |
| 1068 | }); |
| 1069 | const launchedAndRunning = [ |
| 1070 | ...backgroundLaunch(CHILD), |
| 1071 | returnedRunning("call-background"), |
| 1072 | event("session.execution.started", { sessionID: CHILD }), |
| 1073 | ]; |
| 1074 | |
| 1075 | it.effect("settles every call that joined a running subagent with its one report", () => |
| 1076 | Effect.gen(function* () { |
| 1077 | const { runtime, thread } = yield* resumed([ |
| 1078 | ...launchedAndRunning, |
| 1079 | // OpenCode steers the running subagent and joins its run: one report answers both calls. |
| 1080 | ...callAgain({ background: true }), |
| 1081 | event("session.execution.succeeded", { sessionID: SESSION }), |
| 1082 | event("session.execution.succeeded", { sessionID: CHILD }), |
| 1083 | childReport("msg_report", "CHILD_OK"), |
| 1084 | ]); |
| 1085 | const watch = yield* watchCalls(runtime, thread); |
| 1086 | yield* runtime.startTurn(withLineage(thread)); |
| 1087 | yield* watch.until( |
| 1088 | () => |
| 1089 | watch.calls.get("Again")?.status === "completed" && |
| 1090 | watch.calls.get("Sleep")?.status === "completed" && |
| 1091 | watch.state.roster?.length === 0, |
| 1092 | ); |
| 1093 | assert.deepEqual(Object.fromEntries(watch.calls), { |
| 1094 | Sleep: { status: "completed", result: "CHILD_OK" }, |
| 1095 | Again: { status: "completed", result: "CHILD_OK" }, |
| 1096 | }); |
| 1097 | }).pipe(Effect.scoped), |
| 1098 | ); |
| 1099 | |
| 1100 | it.effect("keeps a foreground call that joined a background run for that run's report", () => |
| 1101 | Effect.gen(function* () { |
| 1102 | const { runtime, thread } = yield* resumed([ |
| 1103 | ...launchedAndRunning, |
| 1104 | // Without `background`, the call joins the run in the background all the same. |
| 1105 | ...callAgain({}), |
| 1106 | event("session.execution.succeeded", { sessionID: SESSION }), |
| 1107 | event("session.execution.succeeded", { sessionID: CHILD }), |
| 1108 | childReport("msg_report", "CHILD_OK"), |
| 1109 | ]); |
| 1110 | const watch = yield* watchCalls(runtime, thread); |
| 1111 | yield* runtime.startTurn(withLineage(thread)); |
| 1112 | yield* watch.until( |
| 1113 | () => |
| 1114 | watch.calls.get("Again")?.status === "completed" && |
| 1115 | watch.calls.get("Sleep")?.status === "completed" && |
| 1116 | watch.state.roster?.length === 0, |
| 1117 | ); |
| 1118 | // Not settled empty when the turn that made it ended. |
| 1119 | assert.deepEqual(watch.calls.get("Again"), { status: "completed", result: "CHILD_OK" }); |
| 1120 | }).pipe(Effect.scoped), |
| 1121 | ); |
| 1122 | |
| 1123 | it.effect("keeps a call to a finished subagent for its own run's report", () => |
| 1124 | Effect.gen(function* () { |
| 1125 | const { runtime, thread } = yield* resumed([ |
| 1126 | ...launchedAndRunning, |
| 1127 | event("session.execution.succeeded", { sessionID: CHILD }), |
| 1128 | // Called again after it ended but before its report: the call starts a new run. |
| 1129 | ...callAgain({ background: true }), |
| 1130 | childReport("msg_report_1", "FIRST_OK"), |
| 1131 | event("session.execution.started", { sessionID: CHILD }), |
| 1132 | event("session.execution.succeeded", { sessionID: CHILD }), |
| 1133 | event("session.execution.succeeded", { sessionID: SESSION }), |
| 1134 | childReport("msg_report_2", "SECOND_OK"), |
| 1135 | ]); |
| 1136 | const watch = yield* watchCalls(runtime, thread); |
| 1137 | yield* runtime.startTurn(withLineage(thread)); |
| 1138 | // The new run's turn starts after the first report was handled. |
| 1139 | yield* watch.until(() => watch.childTurns.has(`${CHILD}:turn:2`)); |
| 1140 | assert.deepEqual(watch.callsAtChildTurnStart.get(`${CHILD}:turn:2`), { |
| 1141 | Sleep: { status: "completed", result: "FIRST_OK" }, |
| 1142 | Again: { status: "running", result: null }, |
| 1143 | }); |
| 1144 | yield* watch.until(() => watch.calls.get("Again")?.status === "completed"); |
| 1145 | assert.deepEqual(watch.calls.get("Again"), { status: "completed", result: "SECOND_OK" }); |
| 1146 | }).pipe(Effect.scoped), |
| 1147 | ); |
| 1148 | |
| 956 | 1149 | it.effect("leaves no background work pending once a reconnect finds its subagent gone", () => |
| 957 | 1150 | Effect.gen(function* () { |
| 958 | 1151 | // The subagent ends while the stream is down: its end, its report and |
Nikita Koynov <43469098+nkoynov@users.noreply.github.com> 2026-10-10 15:19:23+03:00
GitHub <noreply@github.com> 2026-10-10 07:19:23-05:00