| author |  Yash Singh <saiansh2525@gmail.com> 2026-10-12 20:26:16-05:00 |
| committer |  GitHub <noreply@github.com> 2026-10-12 20:26:16-05:00 |
| log | 265fb7000f84d12a44c4e87c17426e1f29e0bb36 |
| tree | ad68a44a600967696e1d770e0898dcd87badff94 |
| parent | 5d6e4192bfcf9eca09741605e12c72b5970025e4 |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): apply confirmed Claude subagent outcomes (#18778)
10 files changed, 979 insertions(+), 84 deletions(-)
apps
packages
contracts
provider-core
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts+414-16
| ... | @@ -60,6 +60,7 @@ import * as Option from "effect/Option"; | ... | @@ -60,6 +60,7 @@ import * as Option from "effect/Option"; |
| 60 | import * as Path from "effect/Path"; | 60 | import * as Path from "effect/Path"; |
| 61 | import * as Queue from "effect/Queue"; | 61 | import * as Queue from "effect/Queue"; |
| 62 | import * as Schema from "effect/Schema"; | 62 | import * as Schema from "effect/Schema"; |
| 63 | import * as Scheduler from "effect/Scheduler"; | ||
| 63 | import * as Scope from "effect/Scope"; | 64 | import * as Scope from "effect/Scope"; |
| 64 | import * as Stream from "effect/Stream"; | 65 | import * as Stream from "effect/Stream"; |
| 65 | import * as TestClock from "effect/testing/TestClock"; | 66 | import * as TestClock from "effect/testing/TestClock"; |
| ... | @@ -2317,7 +2318,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -2317,7 +2318,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 2317 | const makeWakeHarnessWithOptions = (options?: { | 2318 | const makeWakeHarnessWithOptions = (options?: { |
| 2318 | readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>; | 2319 | readonly close?: (sdkMessages: Queue.Queue<SDKMessage>) => Effect.Effect<void>; |
| 2319 | readonly interrupt?: Effect.Effect<void>; | 2320 | readonly interrupt?: Effect.Effect<void>; |
| 2320 | readonly stopTask?: (taskId: string) => Effect.Effect<void>; | 2321 | readonly stopTask?: ( |
| 2322 | taskId: string, | ||
| 2323 | ) => Effect.Effect<void, ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError>; | ||
| 2321 | readonly environment?: NodeJS.ProcessEnv; | 2324 | readonly environment?: NodeJS.ProcessEnv; |
| 2322 | readonly subagentLaunchToolUseId?: string; | 2325 | readonly subagentLaunchToolUseId?: string; |
| 2323 | readonly nativeSessionId?: string; | 2326 | readonly nativeSessionId?: string; |
| ... | @@ -8792,17 +8795,373 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -8792,17 +8795,373 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 8792 | ); | 8795 | ); |
| 8793 | 8796 | ||
| 8794 | it.effect.each([false, true])( | 8797 | it.effect.each([false, true])( |
| 8795 | "stops one native subagent while its sibling keeps running, owner settled %s", | 8798 | "drops an idle start superseded by live completion, later resume %s", |
| 8796 | (ownerSettled) => | 8799 | (resume) => |
| 8800 | Effect.scoped( | ||
| 8801 | Effect.gen(function* () { | ||
| 8802 | const harness = yield* makeWakeHarness; | ||
| 8803 | const startTurn = (ordinal: number, continuation = false) => | ||
| 8804 | harness.runtime.startTurn( | ||
| 8805 | makeClaudeTestTurnInput({ | ||
| 8806 | threadId: harness.threadId, | ||
| 8807 | providerThread: harness.providerThread, | ||
| 8808 | now: DateTime.makeUnsafe(0), | ||
| 8809 | attemptId: RunAttemptId.make(`attempt-stale-start-${ordinal}`), | ||
| 8810 | providerTurnOrdinal: ordinal, | ||
| 8811 | text: "Continue work.", | ||
| 8812 | attachments: [], | ||
| 8813 | ...(continuation | ||
| 8814 | ? { | ||
| 8815 | messageCreatedBy: "agent" as const, | ||
| 8816 | messageCreationSource: "provider" as const, | ||
| 8817 | } | ||
| 8818 | : {}), | ||
| 8819 | }), | ||
| 8820 | ); | ||
| 8821 | const started = (taskId: string, suffix: string) => | ||
| 8822 | makeSubagentTaskStartedFrame({ | ||
| 8823 | taskId, | ||
| 8824 | toolUseId: `toolu-${taskId}`, | ||
| 8825 | uuid: `00000000-0000-4000-8000-${suffix}`, | ||
| 8826 | }); | ||
| 8827 | const finished = (taskId: string, suffix: string) => | ||
| 8828 | makeSubagentNotificationFrame({ | ||
| 8829 | taskId, | ||
| 8830 | toolUseId: `toolu-${taskId}`, | ||
| 8831 | summary: "Finished audit.", | ||
| 8832 | uuid: `00000000-0000-4000-8000-${suffix}`, | ||
| 8833 | }); | ||
| 8834 | yield* startTurn(1); | ||
| 8835 | yield* harness.offerAndWait(started("stale-agent", "000000004001")); | ||
| 8836 | yield* harness.offerAndWait(started("wake-agent", "000000004002")); | ||
| 8837 | yield* harness.offerAndWait( | ||
| 8838 | makeResultFrame({ | ||
| 8839 | uuid: "00000000-0000-4000-8000-000000004003", | ||
| 8840 | result: "Agents launched.", | ||
| 8841 | }), | ||
| 8842 | ); | ||
| 8843 | yield* Queue.take(harness.terminalReceipts); | ||
| 8844 | // The background task resumes in the idle gap, before a user turn. | ||
| 8845 | yield* harness.offerAndWait(started("stale-agent", "000000004004")); | ||
| 8846 | yield* startTurn(2); | ||
| 8847 | yield* harness.offerAndWait(finished("stale-agent", "000000004005")); | ||
| 8848 | yield* harness.offerAndWait( | ||
| 8849 | makeResultFrame({ | ||
| 8850 | uuid: "00000000-0000-4000-8000-000000004006", | ||
| 8851 | result: "User turn finished.", | ||
| 8852 | }), | ||
| 8853 | ); | ||
| 8854 | yield* Queue.take(harness.terminalReceipts); | ||
| 8855 | if (resume) yield* harness.offerAndWait(started("stale-agent", "000000004007")); | ||
| 8856 | yield* harness.offerAndWait(finished("wake-agent", "000000004008")); | ||
| 8857 | yield* Queue.take(harness.continuationReceipts); | ||
| 8858 | yield* startTurn(3, true); | ||
| 8859 | yield* harness.offerAndWait( | ||
| 8860 | makeResultFrame({ | ||
| 8861 | uuid: "00000000-0000-4000-8000-000000004009", | ||
| 8862 | result: "Wake finished.", | ||
| 8863 | }), | ||
| 8864 | ); | ||
| 8865 | yield* Queue.take(harness.terminalReceipts); | ||
| 8866 | const latest = harness.events.findLast( | ||
| 8867 | (event) => | ||
| 8868 | event.type === "subagent.updated" && | ||
| 8869 | event.subagent.nativeTaskRef?.nativeId === "stale-agent", | ||
| 8870 | ); | ||
| 8871 | assert.equal(latest?.type, "subagent.updated"); | ||
| 8872 | if (latest?.type !== "subagent.updated") return; | ||
| 8873 | assert.equal(latest.subagent.status, resume ? "running" : "completed"); | ||
| 8874 | assert.equal(latest.subagent.result, resume ? null : "Finished audit."); | ||
| 8875 | assert.equal(latest.subagent.completedAt === null, resume); | ||
| 8876 | }).pipe( | ||
| 8877 | Effect.provide( | ||
| 8878 | Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), | ||
| 8879 | ), | ||
| 8880 | ), | ||
| 8881 | ), | ||
| 8882 | ); | ||
| 8883 | |||
| 8884 | const makeStopFixture = (options?: Parameters<typeof makeWakeHarnessWithOptions>[0]) => | ||
| 8885 | Effect.gen(function* () { | ||
| 8886 | const requests = yield* Queue.unbounded<void>(); | ||
| 8887 | const harness = yield* makeWakeHarnessWithOptions({ | ||
| 8888 | ...options, | ||
| 8889 | stopTask: (id) => | ||
| 8890 | Queue.offer(requests, undefined).pipe( | ||
| 8891 | Effect.andThen(options?.stopTask?.(id) ?? Effect.void), | ||
| 8892 | ), | ||
| 8893 | }); | ||
| 8894 | let sequence = 4500; | ||
| 8895 | const uuid = () => `00000000-0000-4000-8000-${String(++sequence).padStart(12, "0")}`; | ||
| 8896 | const startTurn = (continuation = false) => | ||
| 8897 | harness.runtime.startTurn( | ||
| 8898 | makeClaudeTestTurnInput({ | ||
| 8899 | threadId: harness.threadId, | ||
| 8900 | providerThread: harness.providerThread, | ||
| 8901 | now: DateTime.makeUnsafe(0), | ||
| 8902 | attemptId: RunAttemptId.make(`attempt-stop-${continuation ? 2 : 1}`), | ||
| 8903 | providerTurnOrdinal: continuation ? 2 : 1, | ||
| 8904 | text: "Run an audit.", | ||
| 8905 | attachments: [], | ||
| 8906 | ...(continuation | ||
| 8907 | ? { messageCreatedBy: "agent" as const, messageCreationSource: "provider" as const } | ||
| 8908 | : {}), | ||
| 8909 | }), | ||
| 8910 | ); | ||
| 8911 | const start = (toolUseId = "toolu-stop-agent") => | ||
| 8912 | harness.offerAndWait( | ||
| 8913 | makeSubagentTaskStartedFrame({ | ||
| 8914 | taskId: "stop-agent", | ||
| 8915 | toolUseId, | ||
| 8916 | uuid: uuid(), | ||
| 8917 | }), | ||
| 8918 | ); | ||
| 8919 | const notify = ( | ||
| 8920 | status: "completed" | "failed" | "stopped", | ||
| 8921 | summary = "Native outcome.", | ||
| 8922 | toolUseId = "toolu-stop-agent", | ||
| 8923 | ) => | ||
| 8924 | harness.offerAndWait( | ||
| 8925 | claudeSdkFrame({ | ||
| 8926 | ...makeSubagentNotificationFrame({ | ||
| 8927 | taskId: "stop-agent", | ||
| 8928 | toolUseId, | ||
| 8929 | summary, | ||
| 8930 | uuid: uuid(), | ||
| 8931 | }), | ||
| 8932 | status, | ||
| 8933 | }), | ||
| 8934 | ); | ||
| 8935 | const stop = (requestedAt?: DateTime.Utc) => | ||
| 8936 | harness.runtime.stopSubagent!({ | ||
| 8937 | providerThread: harness.providerThread, | ||
| 8938 | nativeTaskId: "stop-agent", | ||
| 8939 | ...(requestedAt === undefined ? {} : { requestedAt }), | ||
| 8940 | }); | ||
| 8941 | const finishParent = () => | ||
| 8942 | harness | ||
| 8943 | .offerAndWait(makeResultFrame({ uuid: uuid(), result: "Launched." })) | ||
| 8944 | .pipe(Effect.andThen(Queue.take(harness.terminalReceipts))); | ||
| 8945 | yield* startTurn(); | ||
| 8946 | yield* start(); | ||
| 8947 | yield* Queue.take(harness.subagentReceipts); | ||
| 8948 | return { | ||
| 8949 | ...harness, | ||
| 8950 | start, | ||
| 8951 | notify, | ||
| 8952 | stop, | ||
| 8953 | finishParent, | ||
| 8954 | replay: () => startTurn(true), | ||
| 8955 | uuid, | ||
| 8956 | requests, | ||
| 8957 | }; | ||
| 8958 | }); | ||
| 8959 | |||
| 8960 | it.effect.each([ | ||
| 8961 | [false, "no notification"], | ||
| 8962 | [true, "no notification"], | ||
| 8963 | [false, "stop fails"], | ||
| 8964 | [false, "stop hangs"], | ||
| 8965 | ] as const)("never infers an outcome from Stop, owner settled %s and %s", ([idle, failure]) => | ||
| 8966 | Effect.scoped( | ||
| 8967 | Effect.gen(function* () { | ||
| 8968 | const requested = yield* Deferred.make<void>(); | ||
| 8969 | const fixture = yield* makeStopFixture({ | ||
| 8970 | stopTask: () => | ||
| 8971 | Deferred.succeed(requested, undefined).pipe( | ||
| 8972 | Effect.andThen( | ||
| 8973 | failure === "stop fails" | ||
| 8974 | ? new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({ | ||
| 8975 | method: "stopTask", | ||
| 8976 | cause: "Task is no longer live", | ||
| 8977 | }) | ||
| 8978 | : failure === "stop hangs" | ||
| 8979 | ? Effect.never | ||
| 8980 | : Effect.void, | ||
| 8981 | ), | ||
| 8982 | ), | ||
| 8983 | }); | ||
| 8984 | if (idle) yield* fixture.finishParent(); | ||
| 8985 | const stopping = yield* fixture.stop().pipe(Effect.exit, Effect.forkScoped); | ||
| 8986 | yield* Deferred.await(requested); | ||
| 8987 | yield* TestClock.adjust("2 seconds"); | ||
| 8988 | const result = yield* Fiber.join(stopping); | ||
| 8989 | assert.isTrue(Exit.isFailure(result)); | ||
| 8990 | assert.equal( | ||
| 8991 | fixture.events.findLast((event) => event.type === "subagent.updated")?.subagent.status, | ||
| 8992 | "running", | ||
| 8993 | ); | ||
| 8994 | assert.isTrue(yield* fixture.hasPendingBackgroundWork); | ||
| 8995 | yield* fixture.notify("completed"); | ||
| 8996 | assert.equal((yield* Queue.take(fixture.subagentReceipts)).subagent.status, "completed"); | ||
| 8997 | assert.isFalse(yield* fixture.hasPendingBackgroundWork); | ||
| 8998 | }).pipe( | ||
| 8999 | Effect.provide( | ||
| 9000 | Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), | ||
| 9001 | ), | ||
| 9002 | ), | ||
| 9003 | ), | ||
| 9004 | ); | ||
| 9005 | |||
| 9006 | it.effect.each([1, 2, 3, 4, 5, 6, 7, 8])( | ||
| 9007 | "accepts completion during Stop at scheduler step %s", | ||
| 9008 | (pauseAt) => | ||
| 9009 | Effect.scoped( | ||
| 9010 | Effect.gen(function* () { | ||
| 9011 | const fixture = yield* makeStopFixture(); | ||
| 9012 | const resumptions = yield* Queue.unbounded<() => void>(); | ||
| 9013 | const scheduler = yield* Scheduler.Scheduler; | ||
| 9014 | const dispatcher = scheduler.makeDispatcher(); | ||
| 9015 | let checks = 0; | ||
| 9016 | // Pause Stop before its SDK request and deliver the outcome while it is suspended. | ||
| 9017 | const stopping = yield* fixture.stop().pipe( | ||
| 9018 | Effect.provideService(Scheduler.Scheduler, { | ||
| 9019 | executionMode: scheduler.executionMode, | ||
| 9020 | shouldYield: () => ++checks === pauseAt, | ||
| 9021 | makeDispatcher: () => ({ | ||
| 9022 | scheduleTask: (task, priority) => { | ||
| 9023 | if (checks === pauseAt) Queue.offerUnsafe(resumptions, task); | ||
| 9024 | else dispatcher.scheduleTask(task, priority); | ||
| 9025 | }, | ||
| 9026 | flush: () => dispatcher.flush(), | ||
| 9027 | }), | ||
| 9028 | }), | ||
| 9029 | Effect.exit, | ||
| 9030 | Effect.forkScoped, | ||
| 9031 | ); | ||
| 9032 | const resume = yield* Queue.take(resumptions); | ||
| 9033 | yield* fixture.notify("completed"); | ||
| 9034 | const completed = yield* Queue.take(fixture.subagentReceipts); | ||
| 9035 | yield* Effect.sync(resume); | ||
| 9036 | yield* TestClock.adjust("2 seconds"); | ||
| 9037 | assert.deepEqual( | ||
| 9038 | yield* Fiber.join(stopping), | ||
| 9039 | Exit.succeed({ | ||
| 9040 | status: "completed" as const, | ||
| 9041 | completedAt: completed.subagent.completedAt!, | ||
| 9042 | result: "Native outcome.", | ||
| 9043 | }), | ||
| 9044 | ); | ||
| 9045 | }).pipe( | ||
| 9046 | Effect.provide( | ||
| 9047 | Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), | ||
| 9048 | ), | ||
| 9049 | ), | ||
| 9050 | ), | ||
| 9051 | ); | ||
| 9052 | |||
| 9053 | it.effect("retains native confirmation after the synchronous Agent tool result", () => | ||
| 9054 | Effect.scoped( | ||
| 9055 | Effect.gen(function* () { | ||
| 9056 | const fixture = yield* makeStopFixture({ | ||
| 9057 | stopTask: () => Effect.die("Task already completed"), | ||
| 9058 | }); | ||
| 9059 | yield* fixture.notify("completed"); | ||
| 9060 | const completed = yield* Queue.take(fixture.subagentReceipts); | ||
| 9061 | yield* fixture.offerAndWait( | ||
| 9062 | claudeSdkFrame({ | ||
| 9063 | type: "user", | ||
| 9064 | message: { | ||
| 9065 | role: "user", | ||
| 9066 | content: [ | ||
| 9067 | { type: "tool_result", tool_use_id: "toolu-stop-agent", content: "Tool result." }, | ||
| 9068 | ], | ||
| 9069 | }, | ||
| 9070 | parent_tool_use_id: null, | ||
| 9071 | uuid: fixture.uuid(), | ||
| 9072 | session_id: WAKE_NATIVE_SESSION, | ||
| 9073 | }), | ||
| 9074 | ); | ||
| 9075 | assert.equal((yield* Queue.take(fixture.subagentReceipts)).subagent.result, "Tool result."); | ||
| 9076 | assert.deepEqual(yield* fixture.stop(), { | ||
| 9077 | status: "completed", | ||
| 9078 | completedAt: completed.subagent.completedAt!, | ||
| 9079 | result: "Native outcome.", | ||
| 9080 | }); | ||
| 9081 | }).pipe( | ||
| 9082 | Effect.provide( | ||
| 9083 | Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), | ||
| 9084 | ), | ||
| 9085 | ), | ||
| 9086 | ), | ||
| 9087 | ); | ||
| 9088 | |||
| 9089 | it.effect.each(["completed", "failed", "stopped"] as const)( | ||
| 9090 | "applies idle %s once, preserves a genuine resume, and only confirms native outcomes", | ||
| 9091 | (status) => | ||
| 9092 | Effect.scoped( | ||
| 9093 | Effect.gen(function* () { | ||
| 9094 | let stops = 0; | ||
| 9095 | const fixture = yield* makeStopFixture({ | ||
| 9096 | stopTask: () => Effect.sync(() => void stops++), | ||
| 9097 | }); | ||
| 9098 | yield* fixture.finishParent(); | ||
| 9099 | yield* TestClock.adjust("1 second"); | ||
| 9100 | yield* fixture.notify(status); | ||
| 9101 | const completed = yield* Queue.take(fixture.subagentReceipts); | ||
| 9102 | const expected = status === "stopped" ? "cancelled" : status; | ||
| 9103 | assert.equal(completed.subagent.status, expected); | ||
| 9104 | assert.lengthOf(fixture.offeredMessages, 1); | ||
| 9105 | assert.isFalse(yield* fixture.hasPendingBackgroundWork); | ||
| 9106 | assert.deepEqual(yield* fixture.stop(), { | ||
| 9107 | status: expected, | ||
| 9108 | completedAt: completed.subagent.completedAt!, | ||
| 9109 | result: "Native outcome.", | ||
| 9110 | }); | ||
| 9111 | assert.equal(stops, 0); | ||
| 9112 | yield* TestClock.adjust("1 second"); | ||
| 9113 | yield* fixture.start("toolu-resume"); | ||
| 9114 | assert.isUndefined(yield* fixture.stop(completed.subagent.completedAt!)); | ||
| 9115 | assert.equal(stops, 0); | ||
| 9116 | const stopping = yield* fixture.stop().pipe(Effect.exit, Effect.forkScoped); | ||
| 9117 | yield* Queue.take(fixture.requests); | ||
| 9118 | const beforeReplay = fixture.events.length; | ||
| 9119 | yield* fixture.replay(); | ||
| 9120 | const reopened = yield* Queue.take(fixture.subagentReceipts); | ||
| 9121 | assert.equal(reopened.subagent.status, "running"); | ||
| 9122 | assert.isNull(reopened.subagent.completedAt); | ||
| 9123 | assert.isFalse( | ||
| 9124 | fixture.events | ||
| 9125 | .slice(beforeReplay) | ||
| 9126 | .some( | ||
| 9127 | (event) => event.type === "subagent.updated" && event.subagent.status !== "running", | ||
| 9128 | ), | ||
| 9129 | ); | ||
| 9130 | yield* TestClock.adjust("2 seconds"); | ||
| 9131 | assert.isTrue(Exit.isFailure(yield* Fiber.join(stopping))); | ||
| 9132 | assert.equal(stops, 1); | ||
| 9133 | yield* fixture.notify("completed", "Resumed outcome.", "toolu-resume"); | ||
| 9134 | const resumed = yield* Queue.take(fixture.subagentReceipts); | ||
| 9135 | assert.equal(resumed.subagent.result, "Resumed outcome."); | ||
| 9136 | assert.isAbove( | ||
| 9137 | DateTime.toEpochMillis(resumed.subagent.completedAt!), | ||
| 9138 | DateTime.toEpochMillis(completed.subagent.completedAt!), | ||
| 9139 | ); | ||
| 9140 | }).pipe( | ||
| 9141 | Effect.provide( | ||
| 9142 | Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), | ||
| 9143 | ), | ||
| 9144 | ), | ||
| 9145 | ), | ||
| 9146 | ); | ||
| 9147 | |||
| 9148 | it.effect.each([ | ||
| 9149 | [false, "stopped"], | ||
| 9150 | [true, "stopped"], | ||
| 9151 | [false, "completed"], | ||
| 9152 | ] as const)( | ||
| 9153 | "stops one native subagent while its sibling keeps running, owner settled %s, outcome %s", | ||
| 9154 | ([ownerSettled, outcome]) => | ||
| 8797 | Effect.scoped( | 9155 | Effect.scoped( |
| 8798 | Effect.gen(function* () { | 9156 | Effect.gen(function* () { |
| 8799 | const stopped: string[] = []; | 9157 | const stopped: string[] = []; |
| 9158 | const stopRequested = yield* Deferred.make<void>(); | ||
| 8800 | let closes = 0; | 9159 | let closes = 0; |
| 8801 | const harness = yield* makeWakeHarnessWithOptions({ | 9160 | const harness = yield* makeWakeHarnessWithOptions({ |
| 8802 | stopTask: (taskId) => | 9161 | stopTask: (taskId) => |
| 8803 | Effect.sync(() => { | 9162 | Effect.sync(() => { |
| 8804 | stopped.push(taskId); | 9163 | stopped.push(taskId); |
| 8805 | }), | 9164 | }).pipe(Effect.andThen(Deferred.succeed(stopRequested, undefined))), |
| 8806 | interrupt: Effect.die("A child stop must not interrupt its owner"), | 9165 | interrupt: Effect.die("A child stop must not interrupt its owner"), |
| 8807 | close: () => | 9166 | close: () => |
| 8808 | Effect.sync(() => { | 9167 | Effect.sync(() => { |
| ... | @@ -8839,7 +9198,11 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -8839,7 +9198,11 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 8839 | } | 9198 | } |
| 8840 | const stop = harness.runtime.stopSubagent; | 9199 | const stop = harness.runtime.stopSubagent; |
| 8841 | assert.isDefined(stop); | 9200 | assert.isDefined(stop); |
| 8842 | yield* stop!({ providerThread: harness.providerThread, nativeTaskId: "agent-stop" }); | 9201 | const stopping = yield* stop!({ |
| 9202 | providerThread: harness.providerThread, | ||
| 9203 | nativeTaskId: "agent-stop", | ||
| 9204 | }).pipe(Effect.forkScoped); | ||
| 9205 | yield* Deferred.await(stopRequested); | ||
| 8843 | assert.deepEqual(stopped, ["agent-stop"]); | 9206 | assert.deepEqual(stopped, ["agent-stop"]); |
| 8844 | yield* harness.offerAndWait( | 9207 | yield* harness.offerAndWait( |
| 8845 | claudeSdkFrame({ | 9208 | claudeSdkFrame({ |
| ... | @@ -8849,7 +9212,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -8849,7 +9212,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 8849 | summary: "Stopped by user", | 9212 | summary: "Stopped by user", |
| 8850 | uuid: "00000000-0000-4000-8000-000000000804", | 9213 | uuid: "00000000-0000-4000-8000-000000000804", |
| 8851 | }), | 9214 | }), |
| 8852 | status: "stopped", | 9215 | status: outcome, |
| 8853 | }), | 9216 | }), |
| 8854 | ); | 9217 | ); |
| 8855 | if (ownerSettled) { | 9218 | if (ownerSettled) { |
| ... | @@ -8867,13 +9230,17 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -8867,13 +9230,17 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 8867 | }), | 9230 | }), |
| 8868 | ); | 9231 | ); |
| 8869 | } | 9232 | } |
| 9233 | yield* Fiber.join(stopping); | ||
| 8870 | const stoppedEvent = yield* Queue.take(harness.subagentReceipts); | 9234 | const stoppedEvent = yield* Queue.take(harness.subagentReceipts); |
| 8871 | assert.equal(stoppedEvent.subagent.status, "cancelled"); | 9235 | assert.equal( |
| 9236 | stoppedEvent.subagent.status, | ||
| 9237 | outcome === "stopped" ? "cancelled" : "completed", | ||
| 9238 | ); | ||
| 8872 | const tasks = harness.events.filter((event) => event.type === "subagent.updated"); | 9239 | const tasks = harness.events.filter((event) => event.type === "subagent.updated"); |
| 8873 | assert.equal( | 9240 | assert.equal( |
| 8874 | tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-stop") | 9241 | tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-stop") |
| 8875 | ?.subagent.status, | 9242 | ?.subagent.status, |
| 8876 | "cancelled", | 9243 | outcome === "stopped" ? "cancelled" : "completed", |
| 8877 | ); | 9244 | ); |
| 8878 | assert.equal( | 9245 | assert.equal( |
| 8879 | tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-keep") | 9246 | tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-keep") |
| ... | @@ -10404,7 +10771,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -10404,7 +10771,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 10404 | () => events.filter((event) => event.type === "turn.terminal").length === 2, | 10771 | () => events.filter((event) => event.type === "turn.terminal").length === 2, |
| 10405 | "user turn terminal after replace", | 10772 | "user turn terminal after replace", |
| 10406 | ); | 10773 | ); |
| 10407 | assert.isTrue(yield* hasPendingBackgroundWork); | 10774 | // The actual completion was applied while idle, before wake replay. |
| 10775 | assert.isFalse(yield* hasPendingBackgroundWork); | ||
| 10776 | assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); | ||
| 10408 | assert.lengthOf(continuationRequests, 1); | 10777 | assert.lengthOf(continuationRequests, 1); |
| 10409 | 10778 | ||
| 10410 | yield* Queue.offer( | 10779 | yield* Queue.offer( |
| ... | @@ -12700,17 +13069,24 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -12700,17 +13069,24 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 12700 | ); | 13069 | ); |
| 12701 | 13070 | ||
| 12702 | it.effect.each([ | 13071 | it.effect.each([ |
| 12703 | ["api_error", false], | 13072 | ["api_error", false, false], |
| 12704 | ["aborted_streaming", false], | 13073 | ["aborted_streaming", false, false], |
| 12705 | ["api_error", true], | 13074 | ["api_error", true, false], |
| 13075 | ["api_error", false, true], | ||
| 13076 | ["aborted_streaming", false, true], | ||
| 12706 | ] as const)( | 13077 | ] as const)( |
| 12707 | "stops the run's live workflow after root result %s, continuation=%s", | 13078 | "stops the run's live workflow after root result %s, continuation=%s, pending Stop=%s", |
| 12708 | ([terminalReason, continuation]) => | 13079 | ([terminalReason, continuation, pendingStop]) => |
| 12709 | Effect.scoped( | 13080 | Effect.scoped( |
| 12710 | Effect.gen(function* () { | 13081 | Effect.gen(function* () { |
| 12711 | const stoppedTasks: Array<string> = []; | 13082 | const stoppedTasks: Array<string> = []; |
| 13083 | const stopRequested = yield* Deferred.make<void>(); | ||
| 12712 | const harness = yield* makeWorkflowHarnessWithOptions({ | 13084 | const harness = yield* makeWorkflowHarnessWithOptions({ |
| 12713 | stopTask: (taskId) => Effect.sync(() => void stoppedTasks.push(taskId)), | 13085 | stopTask: (taskId) => |
| 13086 | Effect.sync(() => void stoppedTasks.push(taskId)).pipe( | ||
| 13087 | Effect.andThen(Deferred.succeed(stopRequested, undefined)), | ||
| 13088 | Effect.asVoid, | ||
| 13089 | ), | ||
| 12714 | }); | 13090 | }); |
| 12715 | const now = yield* DateTime.now; | 13091 | const now = yield* DateTime.now; |
| 12716 | const turnInput = { | 13092 | const turnInput = { |
| ... | @@ -12723,6 +13099,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -12723,6 +13099,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 12723 | }; | 13099 | }; |
| 12724 | yield* harness.runtime.startTurn(makeClaudeTestTurnInput(turnInput)); | 13100 | yield* harness.runtime.startTurn(makeClaudeTestTurnInput(turnInput)); |
| 12725 | yield* launchWorkflow({ harness, snapshotUuid: "00000000-0000-4000-8000-000000001050" }); | 13101 | yield* launchWorkflow({ harness, snapshotUuid: "00000000-0000-4000-8000-000000001050" }); |
| 13102 | const stopping = pendingStop | ||
| 13103 | ? yield* harness.runtime.stopSubagent!({ | ||
| 13104 | providerThread: harness.providerThread, | ||
| 13105 | nativeTaskId: WORKFLOW_TASK_ID, | ||
| 13106 | }).pipe(Effect.exit, Effect.forkScoped) | ||
| 13107 | : undefined; | ||
| 13108 | if (pendingStop) yield* Deferred.await(stopRequested); | ||
| 12726 | if (continuation) { | 13109 | if (continuation) { |
| 12727 | yield* harness.offerAndWait( | 13110 | yield* harness.offerAndWait( |
| 12728 | makeResultFrame({ | 13111 | makeResultFrame({ |
| ... | @@ -12764,7 +13147,22 @@ describe("ClaudeAdapterV2 background wake turns", () => { | ... | @@ -12764,7 +13147,22 @@ describe("ClaudeAdapterV2 background wake turns", () => { |
| 12764 | // workflow was re-attributed to) stops following it, so it is stopped | 13147 | // workflow was re-attributed to) stops following it, so it is stopped |
| 12765 | // instead of left running out of sight. | 13148 | // instead of left running out of sight. |
| 12766 | const settled = terminalReason === "api_error" ? "failed" : "cancelled"; | 13149 | const settled = terminalReason === "api_error" ? "failed" : "cancelled"; |
| 12767 | assert.deepEqual(stoppedTasks, [WORKFLOW_TASK_ID]); | 13150 | assert.deepEqual( |
| 13151 | stoppedTasks, | ||
| 13152 | pendingStop ? [WORKFLOW_TASK_ID, WORKFLOW_TASK_ID] : [WORKFLOW_TASK_ID], | ||
| 13153 | ); | ||
| 13154 | if (stopping !== undefined) { | ||
| 13155 | // Parent failure assigns a local outcome without a task_notification. | ||
| 13156 | yield* TestClock.adjust("2 seconds"); | ||
| 13157 | assert.isTrue(Exit.isFailure(yield* Fiber.join(stopping))); | ||
| 13158 | // Local parent cleanup still cannot provide a confirmed outcome on retry. | ||
| 13159 | const retry = yield* harness.runtime.stopSubagent!({ | ||
| 13160 | providerThread: harness.providerThread, | ||
| 13161 | nativeTaskId: WORKFLOW_TASK_ID, | ||
| 13162 | }).pipe(Effect.exit, Effect.forkScoped); | ||
| 13163 | yield* TestClock.adjust("2 seconds"); | ||
| 13164 | assert.isTrue(Exit.isFailure(yield* Fiber.join(retry))); | ||
| 13165 | } | ||
| 12768 | assert.isFalse(yield* harness.hasPendingBackgroundWork); | 13166 | assert.isFalse(yield* harness.hasPendingBackgroundWork); |
| 12769 | assert.equal(workflowCoordinatorEvents(harness.events).at(-1)?.subagent.status, settled); | 13167 | assert.equal(workflowCoordinatorEvents(harness.events).at(-1)?.subagent.status, settled); |
| 12770 | for (const index of [1, 2]) { | 13168 | for (const index of [1, 2]) { |
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+155-5
| ... | @@ -2957,6 +2957,8 @@ interface ActiveClaudeProviderRetry { | ... | @@ -2957,6 +2957,8 @@ interface ActiveClaudeProviderRetry { |
| 2957 | 2957 | ||
| 2958 | interface ActiveClaudeSubagent { | 2958 | interface ActiveClaudeSubagent { |
| 2959 | task: OrchestrationV2Subagent; | 2959 | task: OrchestrationV2Subagent; |
| 2960 | // Local parent-turn cleanup can end the card without a native task outcome. | ||
| 2961 | readonly confirmedOutcome: ProviderAdapter.ProviderAdapterV2SubagentOutcome | undefined; | ||
| 2960 | // The launch run's root node. task.parentNodeId is that same node, or the | 2962 | // The launch run's root node. task.parentNodeId is that same node, or the |
| 2961 | // owning subagent's node for a subagent another subagent started. | 2963 | // owning subagent's node for a subagent another subagent started. |
| 2962 | readonly rootNodeId: OrchestrationV2ExecutionNode["id"]; | 2964 | readonly rootNodeId: OrchestrationV2ExecutionNode["id"]; |
| ... | @@ -3497,6 +3499,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -3497,6 +3499,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 3497 | // idle, a running subagent's frames are projected through it at once | 3499 | // idle, a running subagent's frames are projected through it at once |
| 3498 | // instead of waiting in the wake buffer for a continuation run. | 3500 | // instead of waiting in the wake buffer for a continuation run. |
| 3499 | const settledTurnByNativeThread = new Map<string, ActiveClaudeTurnContext>(); | 3501 | const settledTurnByNativeThread = new Map<string, ActiveClaudeTurnContext>(); |
| 3502 | const subagentStopOutcomes = new Map< | ||
| 3503 | string, | ||
| 3504 | Deferred.Deferred<ProviderAdapter.ProviderAdapterV2SubagentOutcome> | ||
| 3505 | >(); | ||
| 3506 | // An idle outcome stays buffered for the parent's wake, but must not | ||
| 3507 | // settle the child again after that child resumes. | ||
| 3508 | const appliedSubagentOutcomes = new WeakSet<SDKMessage>(); | ||
| 3500 | // Subagent frames can precede the task_started that registers their | 3509 | // Subagent frames can precede the task_started that registers their |
| 3501 | // subagent (the same race rememberPendingClaudeSubagentLaunch covers). | 3510 | // subagent (the same race rememberPendingClaudeSubagentLaunch covers). |
| 3502 | // They wait here and replay once task_started registers the owner. | 3511 | // They wait here and replay once task_started registers the owner. |
| ... | @@ -4419,6 +4428,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -4419,6 +4428,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 4419 | const ids = claudeSubagentIds(resume.context, resume.taskId); | 4428 | const ids = claudeSubagentIds(resume.context, resume.taskId); |
| 4420 | const now = yield* DateTime.now; | 4429 | const now = yield* DateTime.now; |
| 4421 | const subagent: ActiveClaudeSubagent = { | 4430 | const subagent: ActiveClaudeSubagent = { |
| 4431 | confirmedOutcome: undefined, | ||
| 4422 | childThread: null, | 4432 | childThread: null, |
| 4423 | task: { | 4433 | task: { |
| 4424 | id: ids.nodeId, | 4434 | id: ids.nodeId, |
| ... | @@ -4805,6 +4815,39 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -4805,6 +4815,39 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 4805 | } | 4815 | } |
| 4806 | }); | 4816 | }); |
| 4807 | 4817 | ||
| 4818 | const confirmedSubagentOutcome = (task: OrchestrationV2Subagent | undefined) => { | ||
| 4819 | if ( | ||
| 4820 | task === undefined || | ||
| 4821 | task.completedAt === null || | ||
| 4822 | (task.status !== "completed" && task.status !== "failed" && task.status !== "cancelled") | ||
| 4823 | ) | ||
| 4824 | return; | ||
| 4825 | return { status: task.status, completedAt: task.completedAt, result: task.result }; | ||
| 4826 | }; | ||
| 4827 | |||
| 4828 | const discardBufferedSubagentStarts = ( | ||
| 4829 | nativeThreadId: string, | ||
| 4830 | taskId: string, | ||
| 4831 | outcome: SDKMessage, | ||
| 4832 | ) => | ||
| 4833 | Ref.update(wakeBuffers, (current) => { | ||
| 4834 | const entry = current.get(nativeThreadId); | ||
| 4835 | if (entry === undefined) return current; | ||
| 4836 | const outcomeIndex = entry.messages.indexOf(outcome); | ||
| 4837 | const messages = entry.messages.filter( | ||
| 4838 | (buffered, index) => | ||
| 4839 | !( | ||
| 4840 | (outcomeIndex === -1 || index < outcomeIndex) && | ||
| 4841 | buffered.type === "system" && | ||
| 4842 | (buffered.subtype === "task_started" || buffered.subtype === "task_progress") && | ||
| 4843 | buffered.task_id === taskId | ||
| 4844 | ), | ||
| 4845 | ); | ||
| 4846 | return messages.length === entry.messages.length | ||
| 4847 | ? current | ||
| 4848 | : new Map(current).set(nativeThreadId, { ...entry, messages }); | ||
| 4849 | }); | ||
| 4850 | |||
| 4808 | const updateClaudeSubagentNode = Effect.fnUntraced(function* (input: { | 4851 | const updateClaudeSubagentNode = Effect.fnUntraced(function* (input: { |
| 4809 | readonly context: ActiveClaudeTurnContext; | 4852 | readonly context: ActiveClaudeTurnContext; |
| 4810 | readonly taskId: string; | 4853 | readonly taskId: string; |
| ... | @@ -4827,6 +4870,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -4827,6 +4870,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 4827 | "running" | "completed" | "failed" | "cancelled" | 4870 | "running" | "completed" | "failed" | "cancelled" |
| 4828 | >; | 4871 | >; |
| 4829 | readonly reopen?: boolean; | 4872 | readonly reopen?: boolean; |
| 4873 | readonly nativeOutcome?: Extract< | ||
| 4874 | SDKMessage, | ||
| 4875 | { readonly type: "system"; readonly subtype: "task_notification" } | ||
| 4876 | >; | ||
| 4830 | readonly onlyIfRunningInRun?: RunId | null; | 4877 | readonly onlyIfRunningInRun?: RunId | null; |
| 4831 | }) { | 4878 | }) { |
| 4832 | // The session registry lets a wake-replay turn (fresh context maps) | 4879 | // The session registry lets a wake-replay turn (fresh context maps) |
| ... | @@ -4993,6 +5040,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -4993,6 +5040,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 4993 | } satisfies OrchestrationV2Subagent; | 5040 | } satisfies OrchestrationV2Subagent; |
| 4994 | const subagent = { | 5041 | const subagent = { |
| 4995 | task, | 5042 | task, |
| 5043 | confirmedOutcome: | ||
| 5044 | input.nativeOutcome !== undefined | ||
| 5045 | ? confirmedSubagentOutcome(task) | ||
| 5046 | : isReopen | ||
| 5047 | ? undefined | ||
| 5048 | : existingSubagent?.confirmedOutcome, | ||
| 4996 | rootNodeId: existingSubagent?.rootNodeId ?? input.context.input.rootNodeId, | 5049 | rootNodeId: existingSubagent?.rootNodeId ?? input.context.input.rootNodeId, |
| 4997 | childThreadId, | 5050 | childThreadId, |
| 4998 | childRootNodeId, | 5051 | childRootNodeId, |
| ... | @@ -5037,6 +5090,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -5037,6 +5090,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 5037 | return [true, new Map(current).set(input.taskId, subagent)] as const; | 5090 | return [true, new Map(current).set(input.taskId, subagent)] as const; |
| 5038 | }); | 5091 | }); |
| 5039 | if (!accepted) return; | 5092 | if (!accepted) return; |
| 5093 | if (input.nativeOutcome !== undefined) appliedSubagentOutcomes.add(input.nativeOutcome); | ||
| 5040 | // A call the subagent left open never returns once it ends. It ends | 5094 | // A call the subagent left open never returns once it ends. It ends |
| 5041 | // first, while the run that owns the child thread still takes its | 5095 | // first, while the run that owns the child thread still takes its |
| 5042 | // events. | 5096 | // events. |
| ... | @@ -5210,6 +5264,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -5210,6 +5264,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 5210 | }, | 5264 | }, |
| 5211 | }); | 5265 | }); |
| 5212 | 5266 | ||
| 5267 | if (input.nativeOutcome !== undefined && subagent.confirmedOutcome !== undefined) { | ||
| 5268 | const outcome = subagentStopOutcomes.get(input.taskId); | ||
| 5269 | if (outcome !== undefined) yield* Deferred.succeed(outcome, subagent.confirmedOutcome); | ||
| 5270 | } | ||
| 5271 | |||
| 5213 | // A completed subagent's result is normally its final assistant | 5272 | // A completed subagent's result is normally its final assistant |
| 5214 | // message, which is already in the child thread when its text was | 5273 | // message, which is already in the child thread when its text was |
| 5215 | // routed there. Failures and cancellations always get the message. | 5274 | // routed there. Failures and cancellations always get the message. |
| ... | @@ -6230,6 +6289,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -6230,6 +6289,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 6230 | const { progress: _staleProgress, ...priorTask } = registered.task; | 6289 | const { progress: _staleProgress, ...priorTask } = registered.task; |
| 6231 | return new Map(current).set(message.task_id, { | 6290 | return new Map(current).set(message.task_id, { |
| 6232 | ...registered, | 6291 | ...registered, |
| 6292 | confirmedOutcome: undefined, | ||
| 6233 | task: { | 6293 | task: { |
| 6234 | ...priorTask, | 6294 | ...priorTask, |
| 6235 | status: "running", | 6295 | status: "running", |
| ... | @@ -6699,6 +6759,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -6699,6 +6759,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 6699 | message, | 6759 | message, |
| 6700 | activeContext: null, | 6760 | activeContext: null, |
| 6701 | }); | 6761 | }); |
| 6762 | // Wake delivery must not delay a known child's actual outcome. | ||
| 6763 | // Record wake evidence first, while the child is still running. | ||
| 6764 | const registered = (yield* Ref.get(sessionSubagentsByTaskId)).get(message.task_id); | ||
| 6765 | const settled = settledTurnByNativeThread.get(liveQuery.nativeThreadId); | ||
| 6766 | const hasBufferedStart = | ||
| 6767 | (yield* Ref.get(wakeBuffers)) | ||
| 6768 | .get(liveQuery.nativeThreadId) | ||
| 6769 | ?.messages.some( | ||
| 6770 | (entry) => | ||
| 6771 | entry.type === "system" && | ||
| 6772 | entry.subtype === "task_started" && | ||
| 6773 | entry.task_id === message.task_id, | ||
| 6774 | ) ?? false; | ||
| 6775 | // Starts and workflow transcripts still need the continuation's | ||
| 6776 | // context. A pending Stop must receive its native outcome now. | ||
| 6777 | const canApplyOutcome = | ||
| 6778 | subagentStopOutcomes.has(message.task_id) || | ||
| 6779 | (!hasBufferedStart && registered?.task.workflow === undefined); | ||
| 6780 | if (registered !== undefined && settled !== undefined && canApplyOutcome) { | ||
| 6781 | yield* discardBufferedSubagentStarts( | ||
| 6782 | liveQuery.nativeThreadId, | ||
| 6783 | message.task_id, | ||
| 6784 | message, | ||
| 6785 | ); | ||
| 6786 | yield* updateClaudeSubagentNode({ | ||
| 6787 | context: settled, | ||
| 6788 | taskId: message.task_id, | ||
| 6789 | ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), | ||
| 6790 | workflowFrame: message, | ||
| 6791 | nativeOutcome: message, | ||
| 6792 | result: message.summary, | ||
| 6793 | status: claudeTaskOutcome(message.status), | ||
| 6794 | onlyIfRunningInRun: registered.task.runId, | ||
| 6795 | }); | ||
| 6796 | } | ||
| 6702 | } else { | 6797 | } else { |
| 6703 | yield* applyBackgroundTaskRosterMessage({ | 6798 | yield* applyBackgroundTaskRosterMessage({ |
| 6704 | nativeThreadId: liveQuery.nativeThreadId, | 6799 | nativeThreadId: liveQuery.nativeThreadId, |
| ... | @@ -7186,6 +7281,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -7186,6 +7281,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 7186 | } | 7281 | } |
| 7187 | 7282 | ||
| 7188 | if (message.type === "system" && message.subtype === "task_notification") { | 7283 | if (message.type === "system" && message.subtype === "task_notification") { |
| 7284 | // A live outcome supersedes earlier idle starts. During replay only | ||
| 7285 | // remove frames before this notification, preserving a later resume. | ||
| 7286 | if ( | ||
| 7287 | !appliedSubagentOutcomes.has(message) && | ||
| 7288 | (yield* Ref.get(sessionSubagentsByTaskId)).has(message.task_id) | ||
| 7289 | ) { | ||
| 7290 | yield* discardBufferedSubagentStarts( | ||
| 7291 | liveQuery.nativeThreadId, | ||
| 7292 | message.task_id, | ||
| 7293 | message, | ||
| 7294 | ); | ||
| 7295 | } | ||
| 7189 | // A wake-replay turn has empty ignoredTaskIds, so opaque-task | 7296 | // A wake-replay turn has empty ignoredTaskIds, so opaque-task |
| 7190 | // tracking (live roster, wake eligibility, or the short-lived | 7297 | // tracking (live roster, wake eligibility, or the short-lived |
| 7191 | // post-buffer replay tombstone) classifies local_bash before any | 7298 | // post-buffer replay tombstone) classifies local_bash before any |
| ... | @@ -7228,13 +7335,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -7228,13 +7335,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 7228 | message, | 7335 | message, |
| 7229 | activeContext: context, | 7336 | activeContext: context, |
| 7230 | }); | 7337 | }); |
| 7231 | if (!wasBackgroundTask && !context.ignoredTaskIds.has(message.task_id)) { | 7338 | if ( |
| 7339 | !wasBackgroundTask && | ||
| 7340 | !context.ignoredTaskIds.has(message.task_id) && | ||
| 7341 | !appliedSubagentOutcomes.has(message) | ||
| 7342 | ) { | ||
| 7232 | yield* updateClaudeSubagentNode({ | 7343 | yield* updateClaudeSubagentNode({ |
| 7233 | context, | 7344 | context, |
| 7234 | taskId: message.task_id, | 7345 | taskId: message.task_id, |
| 7235 | ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), | 7346 | ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), |
| 7236 | // Terminal usage totals arrive here, after the last snapshot. | 7347 | // Terminal usage totals arrive here, after the last snapshot. |
| 7237 | workflowFrame: message, | 7348 | workflowFrame: message, |
| 7349 | nativeOutcome: message, | ||
| 7238 | result: message.summary, | 7350 | result: message.summary, |
| 7239 | status: | 7351 | status: |
| 7240 | message.status === "completed" | 7352 | message.status === "completed" |
| ... | @@ -8973,11 +9085,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -8973,11 +9085,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 8973 | const existing = yield* Ref.get(queryContext); | 9085 | const existing = yield* Ref.get(queryContext); |
| 8974 | const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId; | 9086 | const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId; |
| 8975 | const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); | 9087 | const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); |
| 8976 | if (subagent === undefined || subagent.task.status !== "running") return; | 9088 | if (subagent?.task.threadId !== input.providerThread.appThreadId) { |
| 9089 | return yield* new ProviderAdapter.ProviderAdapterProtocolError({ | ||
| 9090 | driver: CLAUDE_PROVIDER, | ||
| 9091 | detail: "Claude subagent has no known owning task to confirm its outcome.", | ||
| 9092 | }); | ||
| 9093 | } | ||
| 9094 | if ( | ||
| 9095 | input.requestedAt !== undefined && | ||
| 9096 | subagent.task.startedAt !== null && | ||
| 9097 | DateTime.toEpochMillis(subagent.task.startedAt) > | ||
| 9098 | DateTime.toEpochMillis(input.requestedAt) | ||
| 9099 | ) | ||
| 9100 | return; | ||
| 9101 | const confirmed = subagent.confirmedOutcome; | ||
| 9102 | if (confirmed !== undefined) return confirmed; | ||
| 8977 | if ( | 9103 | if ( |
| 8978 | existing === null || | 9104 | existing === null || |
| 8979 | existing.nativeThreadId !== nativeThreadId || | 9105 | existing.nativeThreadId !== nativeThreadId || |
| 8980 | subagent.task.threadId !== input.providerThread.appThreadId || | ||
| 8981 | existing.subagentsFromEarlierProcesses.has(subagent) || | 9106 | existing.subagentsFromEarlierProcesses.has(subagent) || |
| 8982 | existing.query.stopTask === undefined | 9107 | existing.query.stopTask === undefined |
| 8983 | ) { | 9108 | ) { |
| ... | @@ -8986,16 +9111,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( | ... | @@ -8986,16 +9111,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( |
| 8986 | detail: "Claude subagent has no live owning query with task stopping support.", | 9111 | detail: "Claude subagent has no live owning query with task stopping support.", |
| 8987 | }); | 9112 | }); |
| 8988 | } | 9113 | } |
| 8989 | yield* existing.query.stopTask(input.nativeTaskId).pipe( | 9114 | const outcome = |
| 9115 | subagentStopOutcomes.get(input.nativeTaskId) ?? | ||
| 9116 | (yield* Deferred.make<ProviderAdapter.ProviderAdapterV2SubagentOutcome>()); | ||
| 9117 | subagentStopOutcomes.set(input.nativeTaskId, outcome); | ||
| 9118 | const recorded = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); | ||
| 9119 | if ( | ||
| 9120 | recorded?.task.startedAt === subagent.task.startedAt && | ||
| 9121 | recorded.confirmedOutcome !== undefined | ||
| 9122 | ) { | ||
| 9123 | if (subagentStopOutcomes.get(input.nativeTaskId) === outcome) | ||
| 9124 | subagentStopOutcomes.delete(input.nativeTaskId); | ||
| 9125 | return recorded.confirmedOutcome; | ||
| 9126 | } | ||
| 9127 | const confirmedOutcome = yield* existing.query.stopTask(input.nativeTaskId).pipe( | ||
| 9128 | Effect.andThen(Deferred.await(outcome)), | ||
| 9129 | Effect.timeout("2 seconds"), | ||
| 8990 | Effect.mapError( | 9130 | Effect.mapError( |
| 8991 | (cause) => | 9131 | (cause) => |
| 8992 | new ProviderAdapter.ProviderAdapterProtocolError({ | 9132 | new ProviderAdapter.ProviderAdapterProtocolError({ |
| 8993 | driver: CLAUDE_PROVIDER, | 9133 | driver: CLAUDE_PROVIDER, |
| 8994 | detail: "Claude native subagent stop failed.", | 9134 | detail: "Claude subagent Stop was not confirmed by a native outcome.", |
| 8995 | cause, | 9135 | cause, |
| 8996 | }), | 9136 | }), |
| 8997 | ), | 9137 | ), |
| 9138 | Effect.ensuring( | ||
| 9139 | Effect.sync(() => { | ||
| 9140 | if (subagentStopOutcomes.get(input.nativeTaskId) === outcome) | ||
| 9141 | subagentStopOutcomes.delete(input.nativeTaskId); | ||
| 9142 | }), | ||
| 9143 | ), | ||
| 8998 | ); | 9144 | ); |
| 9145 | const current = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); | ||
| 9146 | return current?.task.startedAt === subagent.task.startedAt | ||
| 9147 | ? confirmedOutcome | ||
| 9148 | : undefined; | ||
| 8999 | }), | 9149 | }), |
| 9000 | respondToRuntimeRequest: Effect.fn("ClaudeAdapterV2.respondToRuntimeRequest")( | 9150 | respondToRuntimeRequest: Effect.fn("ClaudeAdapterV2.respondToRuntimeRequest")( |
| 9001 | function* (requestInput) { | 9151 | function* (requestInput) { |
apps/server/src/orchestration-v2/EffectWorker.test.ts+65-25
| ... | @@ -839,31 +839,71 @@ it.effect("settles a delegated child once its restart continuation fails for goo | ... | @@ -839,31 +839,71 @@ it.effect("settles a delegated child once its restart continuation fails for goo |
| 839 | }), | 839 | }), |
| 840 | ); | 840 | ); |
| 841 | 841 | ||
| 842 | it.effect("a native subagent stop does not settle its owner's background work", () => | 842 | it.effect.each(["confirmed", "unconfirmed", "failed"] as const)( |
| 843 | Effect.gen(function* () { | 843 | "a native subagent stop only repairs %s provider evidence", |
| 844 | const now = yield* DateTime.now; | 844 | (outcome) => |
| 845 | const events = yield* Ref.make<ReadonlyArray<string>>([]); | 845 | Effect.gen(function* () { |
| 846 | const layer = layerExecutorFor({ | 846 | const now = yield* DateTime.now; |
| 847 | events, | 847 | const events = yield* Ref.make<ReadonlyArray<string>>([]); |
| 848 | interrupt: (input) => { | 848 | const layer = layerExecutorFor({ |
| 849 | assert.equal(input.subagent?.nativeTaskId, "native-task"); | 849 | events, |
| 850 | return Ref.update(events, (current) => [...current, "stop-child"]); | 850 | interrupt: (input) => { |
| 851 | }, | 851 | assert.equal(input.subagent?.nativeTaskId, "native-task"); |
| 852 | threads: { dispatch: () => Effect.die("The owner's background work must stay running") }, | 852 | return Ref.update(events, (current) => [...current, "stop-child"]).pipe( |
| 853 | }); | 853 | Effect.andThen( |
| 854 | yield* Effect.gen(function* () { | 854 | outcome === "failed" |
| 855 | const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; | 855 | ? new ProviderTurnControlService.ProviderTurnControlError({ |
| 856 | yield* executor.execute({ | 856 | threadId, |
| 857 | ...restartEffect(now, { type: "detach" }), | 857 | providerTurnId, |
| 858 | request: { | 858 | operation: "interrupt", |
| 859 | type: "provider-turn.interrupt", | 859 | cause: "SDK stop failed", |
| 860 | providerSessionId: oldSessionId, | 860 | }) |
| 861 | providerThreadId, | 861 | : Effect.succeed( |
| 862 | providerTurnId, | 862 | outcome === "confirmed" |
| 863 | subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" }, | 863 | ? { |
| 864 | status: "cancelled" as const, | ||
| 865 | completedAt: now, | ||
| 866 | result: "Stopped by Claude", | ||
| 867 | } | ||
| 868 | : undefined, | ||
| 869 | ), | ||
| 870 | ), | ||
| 871 | ); | ||
| 872 | }, | ||
| 873 | threads: { | ||
| 874 | dispatch: (command) => { | ||
| 875 | assert.equal(command.type, "thread.background-work.settle"); | ||
| 876 | if (command.type === "thread.background-work.settle") { | ||
| 877 | assert.equal(command.subagent?.id, "subagent"); | ||
| 878 | assert.deepEqual(command.subagent?.outcome, { | ||
| 879 | status: "cancelled", | ||
| 880 | completedAt: now, | ||
| 881 | result: "Stopped by Claude", | ||
| 882 | }); | ||
| 883 | } | ||
| 884 | return Ref.update(events, (current) => [...current, "settle-child"]).pipe( | ||
| 885 | Effect.as({ sequence: 1, storedEvents: [] }), | ||
| 886 | ); | ||
| 887 | }, | ||
| 864 | }, | 888 | }, |
| 865 | }); | 889 | }); |
| 866 | }).pipe(Effect.provide(layer)); | 890 | const execution = yield* Effect.gen(function* () { |
| 867 | assert.deepEqual(yield* Ref.get(events), ["stop-child"]); | 891 | const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; |
| 868 | }), | 892 | yield* executor.execute({ |
| 893 | ...restartEffect(now, { type: "detach" }), | ||
| 894 | request: { | ||
| 895 | type: "provider-turn.interrupt", | ||
| 896 | providerSessionId: oldSessionId, | ||
| 897 | providerThreadId, | ||
| 898 | providerTurnId, | ||
| 899 | subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" }, | ||
| 900 | }, | ||
| 901 | }); | ||
| 902 | }).pipe(Effect.provide(layer), Effect.exit); | ||
| 903 | assert.equal(Exit.isFailure(execution), outcome === "failed"); | ||
| 904 | assert.deepEqual( | ||
| 905 | yield* Ref.get(events), | ||
| 906 | outcome === "confirmed" ? ["stop-child", "settle-child"] : ["stop-child"], | ||
| 907 | ); | ||
| 908 | }), | ||
| 869 | ); | 909 | ); |
apps/server/src/orchestration-v2/EffectWorker.ts+39-17
| ... | @@ -162,51 +162,73 @@ export const layerExecutor: Layer.Layer< | ... | @@ -162,51 +162,73 @@ export const layerExecutor: Layer.Layer< |
| 162 | }), | 162 | }), |
| 163 | ), | 163 | ), |
| 164 | ); | 164 | ); |
| 165 | case "provider-turn.interrupt": | 165 | case "provider-turn.interrupt": { |
| 166 | const request = effect.request; | ||
| 166 | return providerTurnControl | 167 | return providerTurnControl |
| 167 | .interrupt({ | 168 | .interrupt({ |
| 168 | threadId: effect.threadId, | 169 | threadId: effect.threadId, |
| 169 | providerSessionId: effect.request.providerSessionId, | 170 | providerSessionId: request.providerSessionId, |
| 170 | providerThreadId: effect.request.providerThreadId, | 171 | providerThreadId: request.providerThreadId, |
| 171 | providerTurnId: effect.request.providerTurnId, | 172 | providerTurnId: request.providerTurnId, |
| 172 | ...(effect.request.subagent === undefined | 173 | ...(request.subagent === undefined |
| 173 | ? {} | 174 | ? {} |
| 174 | : { subagent: effect.request.subagent }), | 175 | : { |
| 176 | subagent: { | ||
| 177 | ...request.subagent, | ||
| 178 | requestedAt: DateTime.makeUnsafe(effect.createdAt), | ||
| 179 | }, | ||
| 180 | }), | ||
| 175 | }) | 181 | }) |
| 176 | .pipe( | 182 | .pipe( |
| 183 | Effect.tapError((cause) => | ||
| 184 | request.subagent === undefined | ||
| 185 | ? Effect.void | ||
| 186 | : Effect.logWarning("orchestration-v2.subagent-stop-failed", { | ||
| 187 | effectId: effect.id, | ||
| 188 | cause, | ||
| 189 | }), | ||
| 190 | ), | ||
| 177 | Effect.catch((cause) => | 191 | Effect.catch((cause) => |
| 192 | request.subagent === undefined && | ||
| 178 | isNonRetryableProviderTurnControlFailure( | 193 | isNonRetryableProviderTurnControlFailure( |
| 179 | effect.request.type, | 194 | request.type, |
| 180 | Cause.pretty(Cause.fail(cause)), | 195 | Cause.pretty(Cause.fail(cause)), |
| 181 | ) | 196 | ) |
| 182 | ? Effect.void | 197 | ? Effect.void |
| 183 | : Effect.fail(cause), | 198 | : Effect.fail(cause), |
| 184 | ), | 199 | ), |
| 185 | // The provider has stopped what it still ran and reported it. | 200 | // A targeted Stop repairs only an outcome the provider actually observed. |
| 186 | // Whatever the thread still shows on that provider thread is | 201 | // A full-turn Stop retains its existing process-teardown repair. |
| 187 | // work no process will report on, so the Stop ends it too. | 202 | Effect.andThen((outcome) => |
| 188 | // One Stop can interrupt several provider threads, so the | 203 | request.subagent !== undefined && outcome === undefined |
| 189 | // settle is keyed by effect, not by the Stop command. | ||
| 190 | Effect.andThen( | ||
| 191 | effect.request.subagent !== undefined | ||
| 192 | ? Effect.void | 204 | ? Effect.void |
| 193 | : threads.dispatch({ | 205 | : threads.dispatch({ |
| 194 | type: "thread.background-work.settle", | 206 | type: "thread.background-work.settle", |
| 195 | commandId: CommandId.make(`${effect.id}:background-work-settled`), | 207 | commandId: CommandId.make(`${effect.id}:background-work-settled`), |
| 208 | ...(request.subagent === undefined || outcome === undefined | ||
| 209 | ? {} | ||
| 210 | : { | ||
| 211 | subagent: { | ||
| 212 | id: request.subagent.id, | ||
| 213 | stoppedAt: DateTime.makeUnsafe(effect.createdAt), | ||
| 214 | outcome, | ||
| 215 | }, | ||
| 216 | }), | ||
| 196 | threadId: effect.threadId, | 217 | threadId: effect.threadId, |
| 197 | providerThreadId: effect.request.providerThreadId, | 218 | providerThreadId: request.providerThreadId, |
| 198 | providerTurnId: effect.request.providerTurnId, | 219 | providerTurnId: request.providerTurnId, |
| 199 | }), | 220 | }), |
| 200 | ), | 221 | ), |
| 201 | Effect.mapError( | 222 | Effect.mapError( |
| 202 | (cause) => | 223 | (cause) => |
| 203 | new OrchestrationEffectExecutionError({ | 224 | new OrchestrationEffectExecutionError({ |
| 204 | effectId: effect.id, | 225 | effectId: effect.id, |
| 205 | effectType: effect.request.type, | 226 | effectType: request.type, |
| 206 | cause, | 227 | cause, |
| 207 | }), | 228 | }), |
| 208 | ), | 229 | ), |
| 209 | ); | 230 | ); |
| 231 | } | ||
| 210 | case "provider-turn.steer": | 232 | case "provider-turn.steer": |
| 211 | return providerTurnControl | 233 | return providerTurnControl |
| 212 | .steer({ | 234 | .steer({ |
apps/server/src/orchestration-v2/Orchestrator.ts+55-1
| ... | @@ -8917,6 +8917,60 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -8917,6 +8917,60 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8917 | (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), | 8917 | (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), |
| 8918 | ), | 8918 | ), |
| 8919 | ); | 8919 | ); |
| 8920 | if (command.subagent !== undefined) { | ||
| 8921 | const outcome = command.subagent.outcome; | ||
| 8922 | const subagent = projection.subagents.find((task) => task.id === command.subagent?.id); | ||
| 8923 | if ( | ||
| 8924 | subagent?.origin !== "provider_native" || | ||
| 8925 | !["pending", "running", "waiting"].includes(subagent.status) || | ||
| 8926 | (subagent.startedAt !== null && | ||
| 8927 | DateTime.toEpochMillis(subagent.startedAt) > | ||
| 8928 | DateTime.toEpochMillis(command.subagent.stoppedAt)) | ||
| 8929 | ) | ||
| 8930 | return; | ||
| 8931 | const now = yield* DateTime.now; | ||
| 8932 | const allocateEventId = () => | ||
| 8933 | idAllocator.allocate.event({ | ||
| 8934 | threadId: command.threadId, | ||
| 8935 | commandId: command.commandId, | ||
| 8936 | }); | ||
| 8937 | const ended = yield* endOrphanedNativeSubagent({ | ||
| 8938 | projection, | ||
| 8939 | subagentId: subagent.id, | ||
| 8940 | result: outcome.result, | ||
| 8941 | status: outcome.status, | ||
| 8942 | completedAt: outcome.completedAt, | ||
| 8943 | now, | ||
| 8944 | emitted: yield* Ref.get(events), | ||
| 8945 | allocateEventId, | ||
| 8946 | }).pipe(mapDispatchError(command)); | ||
| 8947 | yield* Ref.update(events, (current) => [...current, ...ended]); | ||
| 8948 | if (subagent.childThreadId !== null) { | ||
| 8949 | const child = yield* projectionStore | ||
| 8950 | .getThreadRecords(subagent.childThreadId, ["nodes", "turnItems"], { | ||
| 8951 | turnItemRunIds: [null], | ||
| 8952 | turnItemStatuses: ["pending", "running", "waiting"], | ||
| 8953 | }) | ||
| 8954 | .pipe( | ||
| 8955 | Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), | ||
| 8956 | mapDispatchError(command), | ||
| 8957 | ); | ||
| 8958 | if (child !== null) { | ||
| 8959 | const childEnded = yield* endRunlessRootTurns({ | ||
| 8960 | threadId: subagent.childThreadId, | ||
| 8961 | providerInstanceId: subagent.providerInstanceId, | ||
| 8962 | projection: child, | ||
| 8963 | status: outcome.status, | ||
| 8964 | completedAt: outcome.completedAt, | ||
| 8965 | now, | ||
| 8966 | emitted: yield* Ref.get(events), | ||
| 8967 | allocateEventId, | ||
| 8968 | }).pipe(mapDispatchError(command)); | ||
| 8969 | yield* Ref.update(events, (current) => [...current, ...childEnded]); | ||
| 8970 | } | ||
| 8971 | } | ||
| 8972 | return; | ||
| 8973 | } | ||
| 8920 | const stoppedRunId = projection.attempts.find( | 8974 | const stoppedRunId = projection.attempts.find( |
| 8921 | (attempt) => attempt.id === stopped.providerTurn?.runAttemptId, | 8975 | (attempt) => attempt.id === stopped.providerTurn?.runAttemptId, |
| 8922 | )?.runId; | 8976 | )?.runId; |
| ... | @@ -9016,7 +9070,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -9016,7 +9070,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 9016 | const providerSessionId = providerThread.providerSessionId; | 9070 | const providerSessionId = providerThread.providerSessionId; |
| 9017 | const nativeTaskId = subagent.nativeTaskRef.nativeId; | 9071 | const nativeTaskId = subagent.nativeTaskRef.nativeId; |
| 9018 | const now = yield* DateTime.now; | 9072 | const now = yield* DateTime.now; |
| 9019 | // Record the request with its effect; the provider notification supplies the outcome. | 9073 | // The effect can repair stale projections only after the provider confirms an outcome. |
| 9020 | yield* emit( | 9074 | yield* emit( |
| 9021 | events, | 9075 | events, |
| 9022 | command, | 9076 | command, |
apps/server/src/orchestration-v2/ProviderTurnControlService.ts+26-5
| ... | @@ -9,12 +9,14 @@ import { | ... | @@ -9,12 +9,14 @@ import { |
| 9 | ThreadId, | 9 | ThreadId, |
| 10 | } from "@t3tools/contracts"; | 10 | } from "@t3tools/contracts"; |
| 11 | import * as Clock from "effect/Clock"; | 11 | import * as Clock from "effect/Clock"; |
| 12 | import type * as DateTime from "effect/DateTime"; | ||
| 12 | import * as Context from "effect/Context"; | 13 | import * as Context from "effect/Context"; |
| 13 | import * as Effect from "effect/Effect"; | 14 | import * as Effect from "effect/Effect"; |
| 14 | import * as Layer from "effect/Layer"; | 15 | import * as Layer from "effect/Layer"; |
| 15 | import * as Option from "effect/Option"; | 16 | import * as Option from "effect/Option"; |
| 16 | import * as Schema from "effect/Schema"; | 17 | import * as Schema from "effect/Schema"; |
| 17 | 18 | ||
| 19 | import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; | ||
| 18 | import * as ProjectionStore from "./ProjectionStore.ts"; | 20 | import * as ProjectionStore from "./ProjectionStore.ts"; |
| 19 | import * as ProviderSessionManager from "./ProviderSessionManager.ts"; | 21 | import * as ProviderSessionManager from "./ProviderSessionManager.ts"; |
| 20 | 22 | ||
| ... | @@ -44,12 +46,19 @@ const isProviderTurnControlError = Schema.is(ProviderTurnControlError); | ... | @@ -44,12 +46,19 @@ const isProviderTurnControlError = Schema.is(ProviderTurnControlError); |
| 44 | 46 | ||
| 45 | export interface ProviderTurnControlServiceV2Shape { | 47 | export interface ProviderTurnControlServiceV2Shape { |
| 46 | readonly interrupt: (input: { | 48 | readonly interrupt: (input: { |
| 47 | readonly subagent?: { readonly id: NodeId; readonly nativeTaskId: string }; | 49 | readonly subagent?: { |
| 50 | readonly id: NodeId; | ||
| 51 | readonly nativeTaskId: string; | ||
| 52 | readonly requestedAt?: DateTime.Utc; | ||
| 53 | }; | ||
| 48 | readonly threadId: ThreadId; | 54 | readonly threadId: ThreadId; |
| 49 | readonly providerSessionId: ProviderSessionId; | 55 | readonly providerSessionId: ProviderSessionId; |
| 50 | readonly providerThreadId: ProviderThreadId; | 56 | readonly providerThreadId: ProviderThreadId; |
| 51 | readonly providerTurnId: ProviderTurnId; | 57 | readonly providerTurnId: ProviderTurnId; |
| 52 | }) => Effect.Effect<void, ProviderTurnControlError>; | 58 | }) => Effect.Effect< |
| 59 | void | ProviderAdapter.ProviderAdapterV2SubagentOutcome, | ||
| 60 | ProviderTurnControlError | ||
| 61 | >; | ||
| 53 | readonly steer: (input: { | 62 | readonly steer: (input: { |
| 54 | readonly threadId: ThreadId; | 63 | readonly threadId: ThreadId; |
| 55 | readonly providerSessionId: ProviderSessionId; | 64 | readonly providerSessionId: ProviderSessionId; |
| ... | @@ -178,7 +187,17 @@ export const layer: Layer.Layer< | ... | @@ -178,7 +187,17 @@ export const layer: Layer.Layer< |
| 178 | const session = Option.isSome(loaded.session) | 187 | const session = Option.isSome(loaded.session) |
| 179 | ? loaded.session | 188 | ? loaded.session |
| 180 | : yield* sessions.get(input.providerSessionId); | 189 | : yield* sessions.get(input.providerSessionId); |
| 181 | if (Option.isNone(session)) return; | 190 | if (Option.isNone(session)) { |
| 191 | if (input.subagent !== undefined) { | ||
| 192 | return yield* new ProviderTurnControlError({ | ||
| 193 | threadId: input.threadId, | ||
| 194 | operation: "interrupt", | ||
| 195 | providerTurnId: input.providerTurnId, | ||
| 196 | cause: "No live provider session can confirm this subagent's outcome.", | ||
| 197 | }); | ||
| 198 | } | ||
| 199 | return; | ||
| 200 | } | ||
| 182 | if (input.subagent !== undefined) { | 201 | if (input.subagent !== undefined) { |
| 183 | if (session.value.stopSubagent === undefined) { | 202 | if (session.value.stopSubagent === undefined) { |
| 184 | return yield* new ProviderTurnControlError({ | 203 | return yield* new ProviderTurnControlError({ |
| ... | @@ -188,11 +207,13 @@ export const layer: Layer.Layer< | ... | @@ -188,11 +207,13 @@ export const layer: Layer.Layer< |
| 188 | cause: "The provider does not support stopping a native subagent.", | 207 | cause: "The provider does not support stopping a native subagent.", |
| 189 | }); | 208 | }); |
| 190 | } | 209 | } |
| 191 | yield* session.value.stopSubagent({ | 210 | return yield* session.value.stopSubagent({ |
| 192 | providerThread: loaded.providerThread, | 211 | providerThread: loaded.providerThread, |
| 193 | nativeTaskId: input.subagent.nativeTaskId, | 212 | nativeTaskId: input.subagent.nativeTaskId, |
| 213 | ...(input.subagent.requestedAt === undefined | ||
| 214 | ? {} | ||
| 215 | : { requestedAt: input.subagent.requestedAt }), | ||
| 194 | }); | 216 | }); |
| 195 | return; | ||
| 196 | } | 217 | } |
| 197 | // A settled turn reaches its adapter too: only the adapter knows | 218 | // A settled turn reaches its adapter too: only the adapter knows |
| 198 | // whether it still runs work for the thread, and each one either | 219 | // whether it still runs work for the thread, and each one either |
apps/server/src/orchestration-v2/ThreadStop.test.ts+172-3
| ... | @@ -18,6 +18,8 @@ import * as Exit from "effect/Exit"; | ... | @@ -18,6 +18,8 @@ import * as Exit from "effect/Exit"; |
| 18 | import * as Layer from "effect/Layer"; | 18 | import * as Layer from "effect/Layer"; |
| 19 | import * as Schema from "effect/Schema"; | 19 | import * as Schema from "effect/Schema"; |
| 20 | import * as SqlClient from "effect/sql/SqlClient"; | 20 | import * as SqlClient from "effect/sql/SqlClient"; |
| 21 | import * as Stream from "effect/Stream"; | ||
| 22 | import * as EventStore from "./EventStore.ts"; | ||
| 21 | import * as TestClock from "effect/testing/TestClock"; | 23 | import * as TestClock from "effect/testing/TestClock"; |
| 22 | 24 | ||
| 23 | import * as SqlitePersistence from "../persistence/Sqlite.ts"; | 25 | import * as SqlitePersistence from "../persistence/Sqlite.ts"; |
| ... | @@ -27,6 +29,8 @@ import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; | ... | @@ -27,6 +29,8 @@ import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; |
| 27 | import * as CommandReceiptStore from "./CommandReceiptStore.ts"; | 29 | import * as CommandReceiptStore from "./CommandReceiptStore.ts"; |
| 28 | import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts"; | 30 | import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts"; |
| 29 | import * as Orchestrator from "./Orchestrator.ts"; | 31 | import * as Orchestrator from "./Orchestrator.ts"; |
| 32 | import * as EffectWorker from "./EffectWorker.ts"; | ||
| 33 | import * as EffectOutbox from "./EffectOutbox.ts"; | ||
| 30 | import * as ProjectionStore from "./ProjectionStore.ts"; | 34 | import * as ProjectionStore from "./ProjectionStore.ts"; |
| 31 | import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; | 35 | import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; |
| 32 | import * as ThreadManagementService from "./ThreadManagementService.ts"; | 36 | import * as ThreadManagementService from "./ThreadManagementService.ts"; |
| ... | @@ -49,6 +53,8 @@ const layerTest = ThreadManagementService.layer.pipe( | ... | @@ -49,6 +53,8 @@ const layerTest = ThreadManagementService.layer.pipe( |
| 49 | Layer.mergeAll( | 53 | Layer.mergeAll( |
| 50 | layerDatabase, | 54 | layerDatabase, |
| 51 | ProjectionStore.layer.pipe(Layer.provide(layerDatabase)), | 55 | ProjectionStore.layer.pipe(Layer.provide(layerDatabase)), |
| 56 | EventStore.layer.pipe(Layer.provide(layerDatabase)), | ||
| 57 | EffectOutbox.layer.pipe(Layer.provide(layerDatabase)), | ||
| 52 | ProviderReplayHarness.layerWithRegistry( | 58 | ProviderReplayHarness.layerWithRegistry( |
| 53 | { name: "thread-stop" }, | 59 | { name: "thread-stop" }, |
| 54 | ProviderAdapterRegistry.layerFromAdapters([adapter]), | 60 | ProviderAdapterRegistry.layerFromAdapters([adapter]), |
| ... | @@ -589,9 +595,9 @@ it.effect("thread.stop marks a turn it cannot interrupt so a late agent watch is | ... | @@ -589,9 +595,9 @@ it.effect("thread.stop marks a turn it cannot interrupt so a late agent watch is |
| 589 | }).pipe(Effect.provide(layerTest)), | 595 | }).pipe(Effect.provide(layerTest)), |
| 590 | ); | 596 | ); |
| 591 | 597 | ||
| 592 | it.effect( | 598 | it.effect.each(["missing session", "completed", "failed", "cancelled"] as const)( |
| 593 | "a native child stop persists only its targeted effect and leaves its owner's queue and watch alone", | 599 | "a native child Stop only settles a confirmed %s outcome and preserves its owner and sibling", |
| 594 | () => | 600 | (providerState) => |
| 595 | Effect.gen(function* () { | 601 | Effect.gen(function* () { |
| 596 | const orchestrator = yield* Orchestrator.OrchestratorV2; | 602 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 597 | const projections = yield* ProjectionStore.ProjectionStoreV2; | 603 | const projections = yield* ProjectionStore.ProjectionStoreV2; |
| ... | @@ -625,6 +631,49 @@ it.effect( | ... | @@ -625,6 +631,49 @@ it.effect( |
| 625 | occurredAt: now, | 631 | occurredAt: now, |
| 626 | payload: subagent, | 632 | payload: subagent, |
| 627 | }); | 633 | }); |
| 634 | const sibling = { | ||
| 635 | ...subagent, | ||
| 636 | id: NodeId.make("node:native-stop-sibling"), | ||
| 637 | childThreadId: null, | ||
| 638 | }; | ||
| 639 | yield* projections.apply({ | ||
| 640 | id: EventId.make("event:native-child-stop:sibling"), | ||
| 641 | type: "subagent.updated", | ||
| 642 | threadId, | ||
| 643 | runId: run.id, | ||
| 644 | nodeId: sibling.id, | ||
| 645 | occurredAt: now, | ||
| 646 | payload: sibling, | ||
| 647 | }); | ||
| 648 | const item = before.turnItems.find( | ||
| 649 | (item) => item.type === "subagent" && item.subagentId === subagent.id, | ||
| 650 | )!; | ||
| 651 | yield* projections.apply({ | ||
| 652 | id: EventId.make("event:native-child-stop:item"), | ||
| 653 | type: "turn-item.updated", | ||
| 654 | threadId, | ||
| 655 | runId: run.id, | ||
| 656 | occurredAt: now, | ||
| 657 | payload: { ...item, status: "running", completedAt: null }, | ||
| 658 | }); | ||
| 659 | const childRoot = { | ||
| 660 | ...before.nodes.find((node) => node.id === run.rootNodeId)!, | ||
| 661 | id: NodeId.make("node:native-stop-child-root"), | ||
| 662 | threadId: subagent.childThreadId!, | ||
| 663 | runId: null, | ||
| 664 | parentNodeId: null, | ||
| 665 | status: "running" as const, | ||
| 666 | countsForRun: false, | ||
| 667 | completedAt: null, | ||
| 668 | }; | ||
| 669 | yield* projections.apply({ | ||
| 670 | id: EventId.make("event:native-child-stop:child-root"), | ||
| 671 | type: "node.updated", | ||
| 672 | threadId: childRoot.threadId, | ||
| 673 | nodeId: childRoot.id, | ||
| 674 | occurredAt: now, | ||
| 675 | payload: childRoot, | ||
| 676 | }); | ||
| 628 | const providerThread = before.providerThreads.find( | 677 | const providerThread = before.providerThreads.find( |
| 629 | (thread) => thread.id === run.providerThreadId, | 678 | (thread) => thread.id === run.providerThreadId, |
| 630 | )!; | 679 | )!; |
| ... | @@ -669,6 +718,16 @@ it.effect( | ... | @@ -669,6 +718,16 @@ it.effect( |
| 669 | }); | 718 | }); |
| 670 | yield* send(threadId, "queued", "queue_after_active"); | 719 | yield* send(threadId, "queued", "queue_after_active"); |
| 671 | const stateBefore = yield* threadState(threadId); | 720 | const stateBefore = yield* threadState(threadId); |
| 721 | const eventStore = yield* EventStore.EventStoreV2; | ||
| 722 | const afterSequence = yield* eventStore.latestSequence({ threadId }); | ||
| 723 | const confirmedStatus = | ||
| 724 | providerState === "completed" || providerState === "failed" || providerState === "cancelled" | ||
| 725 | ? providerState | ||
| 726 | : null; | ||
| 727 | const confirmedOutcome = | ||
| 728 | confirmedStatus === null | ||
| 729 | ? undefined | ||
| 730 | : { status: confirmedStatus, completedAt: now, result: "Actual provider outcome" }; | ||
| 672 | const commandId = CommandId.make("stop-native-child"); | 731 | const commandId = CommandId.make("stop-native-child"); |
| 673 | yield* orchestrator.dispatch({ | 732 | yield* orchestrator.dispatch({ |
| 674 | type: "subagent.stop", | 733 | type: "subagent.stop", |
| ... | @@ -703,5 +762,115 @@ it.effect( | ... | @@ -703,5 +762,115 @@ it.effect( |
| 703 | }), | 762 | }), |
| 704 | ); | 763 | ); |
| 705 | assert.isTrue(Exit.isFailure(unrelated)); | 764 | assert.isTrue(Exit.isFailure(unrelated)); |
| 765 | // The provider outcome predates the delayed projection repair. | ||
| 766 | yield* TestClock.adjust("1 second"); | ||
| 767 | // Only execute Stop's durable effect, with no terminal event ingestion. | ||
| 768 | yield* sql`DELETE FROM orchestration_v2_effect_outbox WHERE command_id != ${commandId}`; | ||
| 769 | if (providerState === "missing session") { | ||
| 770 | const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; | ||
| 771 | yield* worker.drain(); | ||
| 772 | } else { | ||
| 773 | yield* orchestrator.dispatch({ | ||
| 774 | type: "thread.background-work.settle", | ||
| 775 | commandId: CommandId.make("repair-native-child"), | ||
| 776 | threadId, | ||
| 777 | providerThreadId: providerThread.id, | ||
| 778 | providerTurnId, | ||
| 779 | subagent: { id: subagent.id, stoppedAt: now, outcome: confirmedOutcome! }, | ||
| 780 | }); | ||
| 781 | } | ||
| 782 | const settled = yield* orchestrator.getThreadProjection(threadId); | ||
| 783 | const cancelled = settled.subagents.find((task) => task.id === subagent.id)!; | ||
| 784 | if (confirmedOutcome === undefined) { | ||
| 785 | const events = yield* eventStore.read({ threadId, afterSequence }).pipe(Stream.runCollect); | ||
| 786 | assert.isFalse( | ||
| 787 | events.some( | ||
| 788 | ({ event }) => | ||
| 789 | (event.type === "subagent.updated" || | ||
| 790 | event.type === "node.updated" || | ||
| 791 | event.type === "turn-item.updated") && | ||
| 792 | ["completed", "failed", "cancelled", "interrupted"].includes(event.payload.status), | ||
| 793 | ), | ||
| 794 | ); | ||
| 795 | assert.equal(cancelled.status, "running"); | ||
| 796 | assert.equal( | ||
| 797 | (yield* orchestrator.getThreadProjection(childRoot.threadId)).nodes.find( | ||
| 798 | (node) => node.id === childRoot.id, | ||
| 799 | )?.status, | ||
| 800 | "running", | ||
| 801 | ); | ||
| 802 | assert.deepEqual(yield* threadState(threadId), stateBefore); | ||
| 803 | return; | ||
| 804 | } | ||
| 805 | |||
| 806 | // Cached clients resume from this sequence, so all lifecycle repairs | ||
| 807 | // must be stored in the parent stream, not just applied to projections. | ||
| 808 | const repaired = yield* eventStore.read({ threadId, afterSequence }).pipe(Stream.runCollect); | ||
| 809 | const terminal = repaired.flatMap(({ event }) => | ||
| 810 | (event.type === "subagent.updated" || | ||
| 811 | event.type === "node.updated" || | ||
| 812 | event.type === "turn-item.updated") && | ||
| 813 | event.payload.status === confirmedStatus | ||
| 814 | ? [event] | ||
| 815 | : [], | ||
| 816 | ); | ||
| 817 | assert.deepEqual(terminal.map((event) => event.type).sort(), [ | ||
| 818 | "node.updated", | ||
| 819 | "subagent.updated", | ||
| 820 | "turn-item.updated", | ||
| 821 | ]); | ||
| 822 | for (const event of terminal) { | ||
| 823 | assert.equal(event.nodeId, subagent.id); | ||
| 824 | assert.equal( | ||
| 825 | DateTime.toEpochMillis(event.payload.completedAt!), | ||
| 826 | DateTime.toEpochMillis(confirmedOutcome!.completedAt), | ||
| 827 | ); | ||
| 828 | if (event.type === "subagent.updated" || event.type === "turn-item.updated") { | ||
| 829 | assert.isDefined(event.payload.updatedAt); | ||
| 830 | } | ||
| 831 | } | ||
| 832 | |||
| 833 | assert.equal(cancelled.status, confirmedStatus); | ||
| 834 | assert.isNotNull(cancelled.completedAt); | ||
| 835 | assert.equal( | ||
| 836 | DateTime.toEpochMillis(cancelled.completedAt!), | ||
| 837 | DateTime.toEpochMillis(confirmedOutcome!.completedAt), | ||
| 838 | ); | ||
| 839 | assert.equal(cancelled.result, confirmedOutcome!.result); | ||
| 840 | assert.equal( | ||
| 841 | settled.turnItems.find((candidate) => candidate.id === item.id)?.status, | ||
| 842 | confirmedStatus, | ||
| 843 | ); | ||
| 844 | const child = yield* orchestrator.getThreadProjection(childRoot.threadId); | ||
| 845 | assert.equal(child.nodes.find((node) => node.id === childRoot.id)?.status, confirmedStatus); | ||
| 846 | assert.isNotNull(child.nodes.find((node) => node.id === childRoot.id)?.completedAt); | ||
| 847 | assert.equal(settled.subagents.find((task) => task.id === sibling.id)?.status, "running"); | ||
| 848 | assert.deepEqual(yield* threadState(threadId), stateBefore); | ||
| 849 | // An old Stop follow-up cannot cancel a genuine subsequent resume. | ||
| 850 | yield* TestClock.adjust("1 second"); | ||
| 851 | const resumedAt = yield* DateTime.now; | ||
| 852 | yield* projections.apply({ | ||
| 853 | id: EventId.make("event:native-child-stop:resume"), | ||
| 854 | type: "subagent.updated", | ||
| 855 | threadId, | ||
| 856 | runId: run.id, | ||
| 857 | nodeId: subagent.id, | ||
| 858 | occurredAt: resumedAt, | ||
| 859 | payload: { ...cancelled, status: "running", startedAt: resumedAt, completedAt: null }, | ||
| 860 | }); | ||
| 861 | yield* orchestrator.dispatch({ | ||
| 862 | type: "thread.background-work.settle", | ||
| 863 | commandId: CommandId.make("old-stop-after-resume"), | ||
| 864 | threadId, | ||
| 865 | providerThreadId: providerThread.id, | ||
| 866 | providerTurnId, | ||
| 867 | subagent: { id: subagent.id, stoppedAt: now, outcome: confirmedOutcome }, | ||
| 868 | }); | ||
| 869 | assert.equal( | ||
| 870 | (yield* orchestrator.getThreadProjection(threadId)).subagents.find( | ||
| 871 | (task) => task.id === subagent.id, | ||
| 872 | )?.status, | ||
| 873 | "running", | ||
| 874 | ); | ||
| 706 | }).pipe(Effect.provide(layerTest)), | 875 | }).pipe(Effect.provide(layerTest)), |
| 707 | ); | 876 | ); |
packages/contracts/src/orchestrationV2.ts+14-1
| ... | @@ -3193,7 +3193,9 @@ const OrchestrationV2InternalCommand = Schema.Union([ | ... | @@ -3193,7 +3193,9 @@ const OrchestrationV2InternalCommand = Schema.Union([ |
| 3193 | * Follows a Stop once its provider returned: background work the settled | 3193 | * Follows a Stop once its provider returned: background work the settled |
| 3194 | * thread still shows on that provider thread is no longer reported by any | 3194 | * thread still shows on that provider thread is no longer reported by any |
| 3195 | * provider process, so it is marked interrupted. Only the stopped turn's run | 3195 | * provider process, so it is marked interrupted. Only the stopped turn's run |
| 3196 | * and older runs are settled; a later run's work is its own. | 3196 | * and older runs are settled; a later run's work is its own. A subagent target |
| 3197 | * limits repair to that task and its runless child work, using a confirmed | ||
| 3198 | * provider outcome. | ||
| 3197 | */ | 3199 | */ |
| 3198 | Schema.Struct({ | 3200 | Schema.Struct({ |
| 3199 | type: Schema.Literal("thread.background-work.settle"), | 3201 | type: Schema.Literal("thread.background-work.settle"), |
| ... | @@ -3201,6 +3203,17 @@ const OrchestrationV2InternalCommand = Schema.Union([ | ... | @@ -3201,6 +3203,17 @@ const OrchestrationV2InternalCommand = Schema.Union([ |
| 3201 | threadId: ThreadId, | 3203 | threadId: ThreadId, |
| 3202 | providerThreadId: ProviderThreadId, | 3204 | providerThreadId: ProviderThreadId, |
| 3203 | providerTurnId: ProviderTurnId, | 3205 | providerTurnId: ProviderTurnId, |
| 3206 | subagent: Schema.optional( | ||
| 3207 | Schema.Struct({ | ||
| 3208 | id: NodeId, | ||
| 3209 | stoppedAt: Schema.DateTimeUtc, | ||
| 3210 | outcome: Schema.Struct({ | ||
| 3211 | status: Schema.Literals(["completed", "failed", "cancelled"]), | ||
| 3212 | completedAt: Schema.DateTimeUtc, | ||
| 3213 | result: Schema.NullOr(Schema.String), | ||
| 3214 | }), | ||
| 3215 | }), | ||
| 3216 | ), | ||
| 3204 | }), | 3217 | }), |
| 3205 | /** | 3218 | /** |
| 3206 | * Stop for one thread, whatever it is doing: interrupts its running turn, holds its queue, | 3219 | * Stop for one thread, whatever it is doing: interrupts its running turn, holds its queue, |
packages/provider-core/src/server/ProviderAdapter.ts+10-1
| ... | @@ -531,6 +531,13 @@ export interface ProviderAdapterV2McpApps { | ... | @@ -531,6 +531,13 @@ export interface ProviderAdapterV2McpApps { |
| 531 | }) => Effect.Effect<{ readonly contents: ReadonlyArray<unknown> }, ProviderAdapterV2Error>; | 531 | }) => Effect.Effect<{ readonly contents: ReadonlyArray<unknown> }, ProviderAdapterV2Error>; |
| 532 | } | 532 | } |
| 533 | 533 | ||
| 534 | /** A terminal task outcome observed by the provider, not inferred from a stop timeout. */ | ||
| 535 | export type ProviderAdapterV2SubagentOutcome = { | ||
| 536 | readonly status: "completed" | "failed" | "cancelled"; | ||
| 537 | readonly completedAt: NonNullable<OrchestrationV2Subagent["completedAt"]>; | ||
| 538 | readonly result: OrchestrationV2Subagent["result"]; | ||
| 539 | }; | ||
| 540 | |||
| 534 | export interface ProviderAdapterV2SessionRuntime { | 541 | export interface ProviderAdapterV2SessionRuntime { |
| 535 | readonly instanceId: ProviderInstanceId; | 542 | readonly instanceId: ProviderInstanceId; |
| 536 | readonly driver: ProviderDriverKind; | 543 | readonly driver: ProviderDriverKind; |
| ... | @@ -606,7 +613,9 @@ export interface ProviderAdapterV2SessionRuntime { | ... | @@ -606,7 +613,9 @@ export interface ProviderAdapterV2SessionRuntime { |
| 606 | readonly stopSubagent?: (input: { | 613 | readonly stopSubagent?: (input: { |
| 607 | readonly providerThread: OrchestrationV2ProviderThread; | 614 | readonly providerThread: OrchestrationV2ProviderThread; |
| 608 | readonly nativeTaskId: string; | 615 | readonly nativeTaskId: string; |
| 609 | }) => Effect.Effect<void, ProviderAdapterV2Error>; | 616 | /** A delayed Stop must not interrupt a task resumed after this request. */ |
| 617 | readonly requestedAt?: NonNullable<OrchestrationV2Subagent["startedAt"]>; | ||
| 618 | }) => Effect.Effect<void | ProviderAdapterV2SubagentOutcome, ProviderAdapterV2Error>; | ||
| 610 | /** | 619 | /** |
| 611 | * Lets a runtime shared by several app threads unload one provider thread's | 620 | * Lets a runtime shared by several app threads unload one provider thread's |
| 612 | * native state (and its MCP servers) when that app thread detaches, while | 621 | * native state (and its MCP servers) when that app thread detaches, while |
packages/provider-core/src/server/subagentProjection.ts+29-10
| ... | @@ -277,13 +277,15 @@ function writtenIds( | ... | @@ -277,13 +277,15 @@ function writtenIds( |
| 277 | export const endOrphanedNativeSubagent = <E>(input: { | 277 | export const endOrphanedNativeSubagent = <E>(input: { |
| 278 | readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "subagents" | "turnItems">; | 278 | readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "subagents" | "turnItems">; |
| 279 | readonly subagentId: NodeId; | 279 | readonly subagentId: NodeId; |
| 280 | readonly status: "interrupted" | "cancelled"; | 280 | readonly status: "completed" | "failed" | "interrupted" | "cancelled"; |
| 281 | readonly completedAt?: DateTime.Utc; | ||
| 282 | readonly result?: string | null; | ||
| 281 | readonly now: DateTime.Utc; | 283 | readonly now: DateTime.Utc; |
| 282 | readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>; | 284 | readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>; |
| 283 | readonly allocateEventId: () => Effect.Effect<EventId, E>; | 285 | readonly allocateEventId: () => Effect.Effect<EventId, E>; |
| 284 | }) => | 286 | }) => |
| 285 | Effect.gen(function* () { | 287 | Effect.gen(function* () { |
| 286 | const { now, status, subagentId } = input; | 288 | const { now, status, subagentId, completedAt = now } = input; |
| 287 | const subagent = input.projection.subagents.find((entry) => entry.id === subagentId); | 289 | const subagent = input.projection.subagents.find((entry) => entry.id === subagentId); |
| 288 | if (subagent?.origin !== "provider_native") return []; | 290 | if (subagent?.origin !== "provider_native") return []; |
| 289 | const node = input.projection.nodes.find((entry) => entry.id === subagentId); | 291 | const node = input.projection.nodes.find((entry) => entry.id === subagentId); |
| ... | @@ -307,7 +309,13 @@ export const endOrphanedNativeSubagent = <E>(input: { | ... | @@ -307,7 +309,13 @@ export const endOrphanedNativeSubagent = <E>(input: { |
| 307 | id: yield* input.allocateEventId(), | 309 | id: yield* input.allocateEventId(), |
| 308 | type: "subagent.updated", | 310 | type: "subagent.updated", |
| 309 | driver: subagent.driver, | 311 | driver: subagent.driver, |
| 310 | payload: { ...subagent, status, completedAt: now, updatedAt: now }, | 312 | payload: { |
| 313 | ...subagent, | ||
| 314 | status, | ||
| 315 | completedAt, | ||
| 316 | updatedAt: now, | ||
| 317 | ...(input.result === undefined ? {} : { result: input.result }), | ||
| 318 | }, | ||
| 311 | }); | 319 | }); |
| 312 | } | 320 | } |
| 313 | if ( | 321 | if ( |
| ... | @@ -319,7 +327,7 @@ export const endOrphanedNativeSubagent = <E>(input: { | ... | @@ -319,7 +327,7 @@ export const endOrphanedNativeSubagent = <E>(input: { |
| 319 | ...scope, | 327 | ...scope, |
| 320 | id: yield* input.allocateEventId(), | 328 | id: yield* input.allocateEventId(), |
| 321 | type: "node.updated", | 329 | type: "node.updated", |
| 322 | payload: { ...node, status, completedAt: now }, | 330 | payload: { ...node, status, completedAt }, |
| 323 | }); | 331 | }); |
| 324 | } | 332 | } |
| 325 | if ( | 333 | if ( |
| ... | @@ -331,7 +339,13 @@ export const endOrphanedNativeSubagent = <E>(input: { | ... | @@ -331,7 +339,13 @@ export const endOrphanedNativeSubagent = <E>(input: { |
| 331 | ...scope, | 339 | ...scope, |
| 332 | id: yield* input.allocateEventId(), | 340 | id: yield* input.allocateEventId(), |
| 333 | type: "turn-item.updated", | 341 | type: "turn-item.updated", |
| 334 | payload: { ...item, status, completedAt: now, updatedAt: now }, | 342 | payload: { |
| 343 | ...item, | ||
| 344 | status, | ||
| 345 | completedAt, | ||
| 346 | updatedAt: now, | ||
| 347 | ...(input.result === undefined ? {} : { result: input.result }), | ||
| 348 | }, | ||
| 335 | }); | 349 | }); |
| 336 | } | 350 | } |
| 337 | return events; | 351 | return events; |
| ... | @@ -346,13 +360,14 @@ export const endRunlessRootTurns = <E>(input: { | ... | @@ -346,13 +360,14 @@ export const endRunlessRootTurns = <E>(input: { |
| 346 | readonly threadId: ThreadId; | 360 | readonly threadId: ThreadId; |
| 347 | readonly providerInstanceId: ProviderInstanceId; | 361 | readonly providerInstanceId: ProviderInstanceId; |
| 348 | readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "turnItems">; | 362 | readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "turnItems">; |
| 349 | readonly status: "interrupted" | "cancelled"; | 363 | readonly status: "completed" | "failed" | "interrupted" | "cancelled"; |
| 364 | readonly completedAt?: DateTime.Utc; | ||
| 350 | readonly now: DateTime.Utc; | 365 | readonly now: DateTime.Utc; |
| 351 | readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>; | 366 | readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>; |
| 352 | readonly allocateEventId: () => Effect.Effect<EventId, E>; | 367 | readonly allocateEventId: () => Effect.Effect<EventId, E>; |
| 353 | }) => | 368 | }) => |
| 354 | Effect.gen(function* () { | 369 | Effect.gen(function* () { |
| 355 | const { now, status } = input; | 370 | const { now, status, completedAt = now } = input; |
| 356 | const endedNodeIds = writtenIds(input.emitted, "node.updated"); | 371 | const endedNodeIds = writtenIds(input.emitted, "node.updated"); |
| 357 | const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); | 372 | const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); |
| 358 | const scope = { | 373 | const scope = { |
| ... | @@ -377,7 +392,7 @@ export const endRunlessRootTurns = <E>(input: { | ... | @@ -377,7 +392,7 @@ export const endRunlessRootTurns = <E>(input: { |
| 377 | id: yield* input.allocateEventId(), | 392 | id: yield* input.allocateEventId(), |
| 378 | type: "node.updated", | 393 | type: "node.updated", |
| 379 | nodeId: node.id, | 394 | nodeId: node.id, |
| 380 | payload: { ...node, status, completedAt: now }, | 395 | payload: { ...node, status, completedAt }, |
| 381 | }); | 396 | }); |
| 382 | } | 397 | } |
| 383 | for (const item of input.projection.turnItems) { | 398 | for (const item of input.projection.turnItems) { |
| ... | @@ -396,8 +411,12 @@ export const endRunlessRootTurns = <E>(input: { | ... | @@ -396,8 +411,12 @@ export const endRunlessRootTurns = <E>(input: { |
| 396 | nodeId: node.id, | 411 | nodeId: node.id, |
| 397 | payload: { | 412 | payload: { |
| 398 | ...item, | 413 | ...item, |
| 399 | status, | 414 | status: |
| 400 | completedAt: now, | 415 | status === "completed" && |
| 416 | (item.type === "command_execution" || item.type === "dynamic_tool") | ||
| 417 | ? "failed" | ||
| 418 | : status, | ||
| 419 | completedAt, | ||
| 401 | updatedAt: now, | 420 | updatedAt: now, |
| 402 | ...(item.type === "reasoning" || item.type === "assistant_message" | 421 | ...(item.type === "reasoning" || item.type === "assistant_message" |
| 403 | ? { streaming: false } | 422 | ? { streaming: false } |