| ... | ... | @@ -69,6 +69,11 @@ const runtimeMock = { |
| 69 | 69 | abortImplementation: null as |
| 70 | 70 | | ((sessionID: string, signal?: AbortSignal) => Promise<void>) |
| 71 | 71 | | null, |
| 72 | sessionChildrenCalls: [] as string[], |
| 73 | sessionChildrenById: new Map<string, Array<{ id: string }>>(), |
| 74 | sessionChildrenImplementation: null as |
| 75 | | ((sessionID: string) => Promise<Array<{ id: string }>>) |
| 76 | | null, |
| 72 | 77 | closeCalls: [] as string[], |
| 73 | 78 | revertCalls: [] as Array<{ sessionID: string; messageID?: string }>, |
| 74 | 79 | messageCalls: [] as Array<{ sessionID: string; messageID: string }>, |
| ... | ... | @@ -116,6 +121,9 @@ const runtimeMock = { |
| 116 | 121 | this.state.abortCalls.length = 0; |
| 117 | 122 | this.state.abortSignals.length = 0; |
| 118 | 123 | this.state.abortImplementation = null; |
| 124 | this.state.sessionChildrenCalls.length = 0; |
| 125 | this.state.sessionChildrenById.clear(); |
| 126 | this.state.sessionChildrenImplementation = null; |
| 119 | 127 | this.state.closeCalls.length = 0; |
| 120 | 128 | this.state.revertCalls.length = 0; |
| 121 | 129 | this.state.messageCalls.length = 0; |
| ... | ... | @@ -252,6 +260,14 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { |
| 252 | 260 | } |
| 253 | 261 | await runtimeMock.state.abortImplementation?.(sessionID, options?.signal); |
| 254 | 262 | }, |
| 263 | children: async ({ sessionID }: { sessionID: string }) => { |
| 264 | runtimeMock.state.sessionChildrenCalls.push(sessionID); |
| 265 | return { |
| 266 | data: runtimeMock.state.sessionChildrenImplementation |
| 267 | ? await runtimeMock.state.sessionChildrenImplementation(sessionID) |
| 268 | : (runtimeMock.state.sessionChildrenById.get(sessionID) ?? []), |
| 269 | }; |
| 270 | }, |
| 255 | 271 | status: async () => { |
| 256 | 272 | runtimeMock.state.sessionStatusCalls += 1; |
| 257 | 273 | if (runtimeMock.state.sessionStatusImplementation) { |
| ... | ... | @@ -1129,6 +1145,9 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { |
| 1129 | 1145 | it.effect("stops a configured-server session without trying to own server lifecycle", () => |
| 1130 | 1146 | Effect.gen(function* () { |
| 1131 | 1147 | const adapter = yield* OpenCodeAdapter; |
| 1148 | const rootSessionId = "http://127.0.0.1:9999/session"; |
| 1149 | runtimeMock.state.sessionChildrenById.set(rootSessionId, [{ id: "ses_stop_child" }]); |
| 1150 | runtimeMock.state.sessionChildrenById.set("ses_stop_child", [{ id: "ses_stop_grandchild" }]); |
| 1132 | 1151 | yield* adapter.startSession({ |
| 1133 | 1152 | provider: ProviderDriverKind.make("opencode"), |
| 1134 | 1153 | threadId: asThreadId("thread-opencode"), |
| ... | ... | @@ -1138,10 +1157,11 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { |
| 1138 | 1157 | yield* adapter.stopSession(asThreadId("thread-opencode")); |
| 1139 | 1158 | |
| 1140 | 1159 | NodeAssert.deepEqual(runtimeMock.state.startCalls, []); |
| 1141 | | NodeAssert.deepEqual( |
| 1142 | | runtimeMock.state.abortCalls.includes("http://127.0.0.1:9999/session"), |
| 1143 | | true, |
| 1144 | | ); |
| 1160 | NodeAssert.deepEqual(runtimeMock.state.abortCalls, [ |
| 1161 | rootSessionId, |
| 1162 | "ses_stop_child", |
| 1163 | "ses_stop_grandchild", |
| 1164 | ]); |
| 1145 | 1165 | }), |
| 1146 | 1166 | ); |
| 1147 | 1167 | |
| ... | ... | @@ -2999,6 +3019,262 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { |
| 2999 | 3019 | }), |
| 3000 | 3020 | ); |
| 3001 | 3021 | |
| 3022 | it.effect("stops the full OpenCode child tree before it completes the interrupt", () => |
| 3023 | Effect.gen(function* () { |
| 3024 | const adapter = yield* OpenCodeAdapter; |
| 3025 | const threadId = asThreadId("thread-interrupt-child-tree"); |
| 3026 | const parentAbortEvent = promiseWithResolvers<unknown>(); |
| 3027 | const markerEvent = promiseWithResolvers<unknown>(); |
| 3028 | const parentAbortStarted = promiseWithResolvers<void>(); |
| 3029 | const parentAbortRelease = promiseWithResolvers<void>(); |
| 3030 | const childAbortStarted = promiseWithResolvers<void>(); |
| 3031 | const childAbortRelease = promiseWithResolvers<void>(); |
| 3032 | const rootSessionId = "http://127.0.0.1:9999/session"; |
| 3033 | runtimeMock.state.subscribedEvents = [parentAbortEvent.promise, markerEvent.promise]; |
| 3034 | runtimeMock.state.sessionChildrenById.set(rootSessionId, [ |
| 3035 | { id: "ses_child_a" }, |
| 3036 | { id: "ses_child_b" }, |
| 3037 | ]); |
| 3038 | runtimeMock.state.sessionChildrenById.set("ses_child_a", [{ id: "ses_grandchild" }]); |
| 3039 | runtimeMock.state.sessionChildrenById.set("ses_unrelated", [{ id: "ses_unrelated_child" }]); |
| 3040 | runtimeMock.state.abortImplementation = async (sessionID) => { |
| 3041 | if (sessionID === rootSessionId) { |
| 3042 | parentAbortStarted.resolve(undefined); |
| 3043 | await parentAbortRelease.promise; |
| 3044 | } |
| 3045 | if (sessionID === "ses_child_a") { |
| 3046 | childAbortStarted.resolve(undefined); |
| 3047 | await childAbortRelease.promise; |
| 3048 | } |
| 3049 | }; |
| 3050 | |
| 3051 | const markerFiber = yield* adapter.streamEvents.pipe( |
| 3052 | Stream.filter( |
| 3053 | (event) => event.threadId === threadId && event.type === "thread.metadata.updated", |
| 3054 | ), |
| 3055 | Stream.runHead, |
| 3056 | Effect.forkChild, |
| 3057 | ); |
| 3058 | yield* adapter.startSession({ |
| 3059 | provider: ProviderDriverKind.make("opencode"), |
| 3060 | threadId, |
| 3061 | runtimeMode: "full-access", |
| 3062 | }); |
| 3063 | const turn = yield* adapter.sendTurn({ |
| 3064 | threadId, |
| 3065 | input: "Run child agents", |
| 3066 | modelSelection: createModelSelection( |
| 3067 | ProviderInstanceId.make("opencode"), |
| 3068 | "opencode/kimi-k3", |
| 3069 | ), |
| 3070 | }); |
| 3071 | |
| 3072 | const interruptFiber = yield* adapter |
| 3073 | .interruptTurn(threadId, turn.turnId) |
| 3074 | .pipe(Effect.result, Effect.forkChild); |
| 3075 | yield* Effect.promise(() => parentAbortStarted.promise); |
| 3076 | runtimeMock.state.sessionChildrenById.get(rootSessionId)?.push({ id: "ses_late_child" }); |
| 3077 | parentAbortEvent.resolve({ |
| 3078 | id: "evt-parent-aborted", |
| 3079 | type: "session.error", |
| 3080 | properties: { |
| 3081 | sessionID: rootSessionId, |
| 3082 | error: { name: "MessageAbortedError", data: { message: "Aborted" } }, |
| 3083 | }, |
| 3084 | }); |
| 3085 | markerEvent.resolve({ |
| 3086 | id: "evt-after-parent-abort", |
| 3087 | type: "session.updated", |
| 3088 | properties: { info: { id: rootSessionId, title: "Parent abort received" } }, |
| 3089 | }); |
| 3090 | yield* Fiber.join(markerFiber); |
| 3091 | |
| 3092 | NodeAssert.equal(interruptFiber.pollUnsafe(), undefined); |
| 3093 | yield* Effect.promise(() => childAbortStarted.promise); |
| 3094 | NodeAssert.equal(interruptFiber.pollUnsafe(), undefined); |
| 3095 | NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_unrelated"), false); |
| 3096 | NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_unrelated_child"), false); |
| 3097 | const sessionsDuringCleanup = yield* adapter.listSessions(); |
| 3098 | const sessionDuringCleanup = sessionsDuringCleanup.find( |
| 3099 | (candidate) => candidate.threadId === threadId, |
| 3100 | ); |
| 3101 | NodeAssert.equal(sessionDuringCleanup?.status, "running"); |
| 3102 | NodeAssert.equal(sessionDuringCleanup?.activeTurnId, turn.turnId); |
| 3103 | const nextTurnFiber = yield* adapter |
| 3104 | .sendTurn({ |
| 3105 | threadId, |
| 3106 | input: "Start after every child stops", |
| 3107 | modelSelection: createModelSelection( |
| 3108 | ProviderInstanceId.make("opencode"), |
| 3109 | "opencode/kimi-k3", |
| 3110 | ), |
| 3111 | }) |
| 3112 | .pipe(Effect.forkChild); |
| 3113 | yield* Effect.yieldNow; |
| 3114 | NodeAssert.equal(runtimeMock.state.promptCalls.length, 1); |
| 3115 | |
| 3116 | childAbortRelease.resolve(undefined); |
| 3117 | parentAbortRelease.resolve(undefined); |
| 3118 | const result = yield* Fiber.join(interruptFiber); |
| 3119 | const nextTurn = yield* Fiber.join(nextTurnFiber); |
| 3120 | NodeAssert.equal(result._tag, "Success"); |
| 3121 | NodeAssert.notEqual(nextTurn.turnId, turn.turnId); |
| 3122 | NodeAssert.equal(runtimeMock.state.promptCalls.length, 2); |
| 3123 | NodeAssert.equal(runtimeMock.state.abortCalls[0], rootSessionId); |
| 3124 | NodeAssert.deepEqual( |
| 3125 | new Set(runtimeMock.state.abortCalls.slice(1)), |
| 3126 | new Set(["ses_child_a", "ses_child_b", "ses_grandchild", "ses_late_child"]), |
| 3127 | ); |
| 3128 | NodeAssert.deepEqual( |
| 3129 | new Set(runtimeMock.state.sessionChildrenCalls), |
| 3130 | new Set([rootSessionId, "ses_child_a", "ses_child_b", "ses_grandchild", "ses_late_child"]), |
| 3131 | ); |
| 3132 | const sessions = yield* adapter.listSessions(); |
| 3133 | const session = sessions.find((candidate) => candidate.threadId === threadId); |
| 3134 | NodeAssert.equal(session?.status, "running"); |
| 3135 | NodeAssert.equal(session?.activeTurnId, nextTurn.turnId); |
| 3136 | |
| 3137 | runtimeMock.state.abortImplementation = null; |
| 3138 | yield* adapter.stopSession(threadId); |
| 3139 | }), |
| 3140 | ); |
| 3141 | |
| 3142 | it.effect("limits SDK requests across the full OpenCode child tree", () => |
| 3143 | Effect.gen(function* () { |
| 3144 | const adapter = yield* OpenCodeAdapter; |
| 3145 | const threadId = asThreadId("thread-interrupt-child-request-limit"); |
| 3146 | const rootSessionId = "http://127.0.0.1:9999/session"; |
| 3147 | const requestRelease = promiseWithResolvers<void>(); |
| 3148 | const limitReached = promiseWithResolvers<void>(); |
| 3149 | let inFlight = 0; |
| 3150 | let maxInFlight = 0; |
| 3151 | const holdRequest = async <T>(result: T): Promise<T> => { |
| 3152 | inFlight += 1; |
| 3153 | maxInFlight = Math.max(maxInFlight, inFlight); |
| 3154 | if (inFlight === 8) { |
| 3155 | limitReached.resolve(undefined); |
| 3156 | } |
| 3157 | await requestRelease.promise; |
| 3158 | inFlight -= 1; |
| 3159 | return result; |
| 3160 | }; |
| 3161 | |
| 3162 | const children = Array.from({ length: 8 }, (_, index) => ({ id: `ses_child_${index}` })); |
| 3163 | runtimeMock.state.sessionChildrenById.set(rootSessionId, children); |
| 3164 | for (const child of children.slice(1)) { |
| 3165 | runtimeMock.state.sessionChildrenById.set( |
| 3166 | child.id, |
| 3167 | Array.from({ length: 8 }, (_, index) => ({ id: `${child.id}_nested_${index}` })), |
| 3168 | ); |
| 3169 | } |
| 3170 | runtimeMock.state.abortImplementation = async (sessionID) => { |
| 3171 | if (sessionID.includes("_nested_")) { |
| 3172 | await holdRequest(undefined); |
| 3173 | } |
| 3174 | }; |
| 3175 | runtimeMock.state.sessionChildrenImplementation = async (sessionID) => { |
| 3176 | if (sessionID === "ses_child_0") { |
| 3177 | return await holdRequest([]); |
| 3178 | } |
| 3179 | return runtimeMock.state.sessionChildrenById.get(sessionID) ?? []; |
| 3180 | }; |
| 3181 | |
| 3182 | yield* adapter.startSession({ |
| 3183 | provider: ProviderDriverKind.make("opencode"), |
| 3184 | threadId, |
| 3185 | runtimeMode: "full-access", |
| 3186 | }); |
| 3187 | const turn = yield* adapter.sendTurn({ |
| 3188 | threadId, |
| 3189 | input: "Run a nested child tree", |
| 3190 | modelSelection: createModelSelection( |
| 3191 | ProviderInstanceId.make("opencode"), |
| 3192 | "opencode/kimi-k3", |
| 3193 | ), |
| 3194 | }); |
| 3195 | |
| 3196 | const interruptFiber = yield* adapter |
| 3197 | .interruptTurn(threadId, turn.turnId) |
| 3198 | .pipe(Effect.forkChild); |
| 3199 | yield* Effect.promise(() => limitReached.promise); |
| 3200 | yield* Effect.yieldNow; |
| 3201 | |
| 3202 | NodeAssert.equal(inFlight, 8); |
| 3203 | NodeAssert.equal(maxInFlight, 8); |
| 3204 | |
| 3205 | requestRelease.resolve(undefined); |
| 3206 | yield* Fiber.join(interruptFiber); |
| 3207 | |
| 3208 | runtimeMock.state.abortImplementation = null; |
| 3209 | runtimeMock.state.sessionChildrenImplementation = null; |
| 3210 | runtimeMock.state.sessionChildrenById.clear(); |
| 3211 | yield* adapter.stopSession(threadId); |
| 3212 | }), |
| 3213 | ); |
| 3214 | |
| 3215 | it.effect("attempts every child abort and fails the interrupt when one child abort fails", () => |
| 3216 | Effect.gen(function* () { |
| 3217 | const adapter = yield* OpenCodeAdapter; |
| 3218 | const threadId = asThreadId("thread-interrupt-child-failure"); |
| 3219 | const rootSessionId = "http://127.0.0.1:9999/session"; |
| 3220 | const failingChildStarted = promiseWithResolvers<void>(); |
| 3221 | const failingChildRelease = promiseWithResolvers<void>(); |
| 3222 | const siblingAbortStarted = promiseWithResolvers<void>(); |
| 3223 | runtimeMock.state.sessionChildrenById.set(rootSessionId, [ |
| 3224 | { id: "ses_failing_child" }, |
| 3225 | { id: "ses_surviving_sibling" }, |
| 3226 | ]); |
| 3227 | runtimeMock.state.abortImplementation = async (sessionID) => { |
| 3228 | if (sessionID === "ses_failing_child") { |
| 3229 | failingChildStarted.resolve(undefined); |
| 3230 | await failingChildRelease.promise; |
| 3231 | throw new Error("child abort failed"); |
| 3232 | } |
| 3233 | if (sessionID === "ses_surviving_sibling") { |
| 3234 | siblingAbortStarted.resolve(undefined); |
| 3235 | } |
| 3236 | }; |
| 3237 | |
| 3238 | yield* adapter.startSession({ |
| 3239 | provider: ProviderDriverKind.make("opencode"), |
| 3240 | threadId, |
| 3241 | runtimeMode: "full-access", |
| 3242 | }); |
| 3243 | const turn = yield* adapter.sendTurn({ |
| 3244 | threadId, |
| 3245 | input: "Run child agents", |
| 3246 | modelSelection: createModelSelection( |
| 3247 | ProviderInstanceId.make("opencode"), |
| 3248 | "opencode/kimi-k3", |
| 3249 | ), |
| 3250 | }); |
| 3251 | |
| 3252 | const interruptFiber = yield* adapter |
| 3253 | .interruptTurn(threadId, turn.turnId) |
| 3254 | .pipe(Effect.result, Effect.forkChild); |
| 3255 | yield* Effect.promise(() => failingChildStarted.promise); |
| 3256 | yield* Effect.promise(() => siblingAbortStarted.promise); |
| 3257 | NodeAssert.equal(interruptFiber.pollUnsafe(), undefined); |
| 3258 | failingChildRelease.resolve(undefined); |
| 3259 | const result = yield* Fiber.join(interruptFiber); |
| 3260 | |
| 3261 | NodeAssert.equal(result._tag, "Failure"); |
| 3262 | if (result._tag === "Failure") { |
| 3263 | NodeAssert.equal(result.failure._tag, "ProviderAdapterRequestError"); |
| 3264 | NodeAssert.equal(result.failure.detail, "child abort failed"); |
| 3265 | } |
| 3266 | NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_failing_child"), true); |
| 3267 | NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_surviving_sibling"), true); |
| 3268 | const sessions = yield* adapter.listSessions(); |
| 3269 | const session = sessions.find((candidate) => candidate.threadId === threadId); |
| 3270 | NodeAssert.equal(session?.status, "running"); |
| 3271 | NodeAssert.equal(session?.activeTurnId, turn.turnId); |
| 3272 | |
| 3273 | runtimeMock.state.abortImplementation = null; |
| 3274 | yield* adapter.stopSession(threadId); |
| 3275 | }), |
| 3276 | ); |
| 3277 | |
| 3002 | 3278 | it.effect("keeps an idle event from completing a turn while its abort request is pending", () => |
| 3003 | 3279 | Effect.gen(function* () { |
| 3004 | 3280 | const adapter = yield* OpenCodeAdapter; |
| ... | ... | @@ -4278,11 +4554,20 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { |
| 4278 | 4554 | const threadId = asThreadId("thread-interrupt-provider-error"); |
| 4279 | 4555 | const errorEvent = promiseWithResolvers<unknown>(); |
| 4280 | 4556 | const abortStarted = promiseWithResolvers<void>(); |
| 4281 | | const abortRelease = promiseWithResolvers<void>(); |
| 4557 | const childAbortStarted = promiseWithResolvers<void>(); |
| 4558 | const childAbortRelease = promiseWithResolvers<void>(); |
| 4559 | const rootSessionId = "http://127.0.0.1:9999/session"; |
| 4282 | 4560 | runtimeMock.state.subscribedEvents = [errorEvent.promise]; |
| 4283 | | runtimeMock.state.abortImplementation = async () => { |
| 4284 | | abortStarted.resolve(undefined); |
| 4285 | | await abortRelease.promise; |
| 4561 | runtimeMock.state.sessionChildrenById.set(rootSessionId, [{ id: "ses_error_child" }]); |
| 4562 | runtimeMock.state.abortImplementation = async (sessionID) => { |
| 4563 | if (sessionID === rootSessionId) { |
| 4564 | abortStarted.resolve(undefined); |
| 4565 | await new Promise<void>(() => {}); |
| 4566 | } |
| 4567 | if (sessionID === "ses_error_child") { |
| 4568 | childAbortStarted.resolve(undefined); |
| 4569 | await childAbortRelease.promise; |
| 4570 | } |
| 4286 | 4571 | }; |
| 4287 | 4572 | |
| 4288 | 4573 | const eventsFiber = yield* adapter.streamEvents.pipe( |
| ... | ... | @@ -4313,16 +4598,14 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { |
| 4313 | 4598 | id: "evt-provider-error-after-stop", |
| 4314 | 4599 | type: "session.error", |
| 4315 | 4600 | properties: { |
| 4316 | | sessionID: "http://127.0.0.1:9999/session", |
| 4601 | sessionID: rootSessionId, |
| 4317 | 4602 | error: { |
| 4318 | 4603 | name: "APIError", |
| 4319 | 4604 | data: { message: "Upstream failed", isRetryable: false }, |
| 4320 | 4605 | }, |
| 4321 | 4606 | }, |
| 4322 | 4607 | }); |
| 4323 | | yield* Effect.yieldNow; |
| 4324 | | abortRelease.resolve(undefined); |
| 4325 | | yield* Fiber.join(interruptFiber); |
| 4608 | yield* Effect.promise(() => childAbortStarted.promise); |
| 4326 | 4609 | |
| 4327 | 4610 | const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); |
| 4328 | 4611 | NodeAssert.deepEqual( |
| ... | ... | @@ -4341,7 +4624,41 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { |
| 4341 | 4624 | failed?.type === "turn.completed" ? failed.payload.state : undefined, |
| 4342 | 4625 | "failed", |
| 4343 | 4626 | ); |
| 4627 | const sessionsDuringCleanup = yield* adapter.listSessions(); |
| 4628 | const sessionDuringCleanup = sessionsDuringCleanup.find( |
| 4629 | (candidate) => candidate.threadId === threadId, |
| 4630 | ); |
| 4631 | NodeAssert.equal(sessionDuringCleanup?.status, "error"); |
| 4632 | NodeAssert.equal(sessionDuringCleanup?.activeTurnId, undefined); |
| 4633 | |
| 4634 | const secondInterruptFiber = yield* adapter.interruptTurn(threadId).pipe(Effect.forkChild); |
| 4635 | const nextTurnFiber = yield* adapter |
| 4636 | .sendTurn({ |
| 4637 | threadId, |
| 4638 | input: "Start after child cleanup", |
| 4639 | modelSelection: createModelSelection( |
| 4640 | ProviderInstanceId.make("opencode"), |
| 4641 | "opencode/kimi-k3", |
| 4642 | ), |
| 4643 | }) |
| 4644 | .pipe(Effect.forkChild); |
| 4645 | yield* Effect.yieldNow; |
| 4646 | NodeAssert.equal( |
| 4647 | runtimeMock.state.abortCalls.filter((sessionID) => sessionID === rootSessionId).length, |
| 4648 | 1, |
| 4649 | ); |
| 4650 | NodeAssert.equal(secondInterruptFiber.pollUnsafe(), undefined); |
| 4651 | NodeAssert.equal(nextTurnFiber.pollUnsafe(), undefined); |
| 4652 | NodeAssert.equal(runtimeMock.state.promptCalls.length, 1); |
| 4344 | 4653 | |
| 4654 | childAbortRelease.resolve(undefined); |
| 4655 | yield* Fiber.join(interruptFiber); |
| 4656 | yield* Fiber.join(secondInterruptFiber); |
| 4657 | const nextTurn = yield* Fiber.join(nextTurnFiber); |
| 4658 | NodeAssert.notEqual(nextTurn.turnId, turn.turnId); |
| 4659 | NodeAssert.equal(runtimeMock.state.promptCalls.length, 2); |
| 4660 | |
| 4661 | runtimeMock.state.abortImplementation = null; |
| 4345 | 4662 | yield* adapter.stopSession(threadId); |
| 4346 | 4663 | }), |
| 4347 | 4664 | ); |