# Files changed · PR \#23886 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/23886/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [fix\(server\): Stop stops a thread's subagents again](https://git.cafe/pingdotgg/t3code/pull/23886.md) Version: 16 Head commit: 4cae03c1816552936fb38c665c9de44aaa2414e5 Base commit: d7a8d6f32001444e013aa07c8a468ce3e987f9cf Comparison base: d7a8d6f32001444e013aa07c8a468ce3e987f9cf Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/mcp/OrchestratorMcpService.ts Status: modified; +1; −0 ```diff @@ -2385,6 +2385,7 @@ type: "thread.stop", commandId, threadId: current.childThreadId, + stopDelegatedTasks: false, ...(reason === undefined ? {} : { reason }), }) .pipe( ``` ## apps/server/src/orchestration-v2/EventSink.ts Status: modified; +6; −1 ```diff @@ -661,7 +661,12 @@ commandId: input.commandId, events: normalized, }); + // A Stop that changes nothing can still queue the stop of its delegated tasks. + const sequence = + storedEvents.at(-1)?.sequence ?? + (input.effects.length === 0 + ? undefined + : yield* eventStore.latestSequence({ threadId: input.threadId })); - const sequence = storedEvents.at(-1)?.sequence; if (sequence === undefined) { return yield* Effect.die( new Error(`Command ${input.commandId} produced no orchestration events.`), ``` ## apps/server/src/orchestration-v2/Orchestrator.ts Status: modified; +28; −10 ```diff @@ -535,6 +535,22 @@ }; } +function delegatedTasksStopEffect(command: { + readonly commandId: CommandId; + readonly threadId: ThreadId; + readonly reason?: string | undefined; +}): PendingOrchestrationEffectV2 { + return { + id: `effect:${command.commandId}:delegated-tasks.stop`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "delegated-tasks.stop", + ...(command.reason === undefined ? {} : { reason: command.reason }), + }, + }; +} + const WORKSPACE_PREPARATION_INPUT = "Preparing workspace"; function withoutPreparationResult( @@ -10150,6 +10166,14 @@ ), ); if (projection.thread.deletedAt !== null) return undefined; + // Queued after the interrupt, since a thread's effects run in order. + const stopDelegatedTasks = + command.stopDelegatedTasks === false || + !projection.subagents.some( + (task) => task.origin === "app_owned" && task.childThreadId !== null, + ) + ? Effect.void + : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]); const latestRun = projection.runs.findLast((run) => run.status !== "queued"); const target = projection.runs.findLast( @@ -10189,6 +10213,7 @@ yield* Ref.update(events, (existing) => [...existing, ...written]); yield* Ref.update(effects, (existing) => [...existing, ...enqueued]); yield* cancelIdleOrphans({ command, events, projection }); + yield* stopDelegatedTasks; return interrupted.value; } yield* Effect.logWarning("thread.stop could not interrupt the running turn", { @@ -10209,6 +10234,7 @@ } yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now }); yield* cancelIdleOrphans({ command, events, projection }); + yield* stopDelegatedTasks; return undefined; }); @@ -11509,15 +11535,7 @@ }); yield* Ref.update(effects, (existing) => [ ...existing, + delegatedTasksStopEffect(command), - { - id: `effect:${command.commandId}:delegated-tasks.stop`, - commandId: command.commandId, - threadId: command.threadId, - request: { - type: "delegated-tasks.stop", - ...(command.reason === undefined ? {} : { reason: command.reason }), - }, - } satisfies PendingOrchestrationEffectV2, ]); } break; @@ -11787,7 +11805,7 @@ ), ); - if (plan.events.length === 0) { + if (plan.events.length === 0 && plan.effects.length === 0) { // A settle that ended nothing still records its receipt: a replayed Stop // effect then finds it instead of settling work that appeared since. const resultSequence = yield* Effect.gen(function* () { ``` ## apps/server/src/orchestration-v2/ThreadManagementService.ts Status: modified; +1; −0 ```diff @@ -1115,6 +1115,7 @@ type: "thread.stop", commandId: CommandId.make(`${input.commandId}:stop:${threadId}`), threadId, + stopDelegatedTasks: false, ...(input.reason === undefined ? {} : { reason: input.reason }), }).pipe( Effect.andThen(stopDelegatedTasks({ ...input, threadId })), ``` ## apps/server/src/orchestration-v2/ThreadStop.test.ts Status: modified; +99; −40 ```diff @@ -530,65 +530,124 @@ }).pipe(Effect.provide(layerTest)), ); +it.effect.each(["run.interrupt", "thread.stop"] as const)( + "%s Stop ends watches, holds queues, and stops the delegated tasks under the thread", + (type) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threads = yield* ThreadManagementService.ThreadManagementService; + const sql = yield* SqlClient.SqlClient; + const parentThreadId = ThreadId.make("thread:stop-parent"); + yield* createWatchingThread(parentThreadId, 1); + yield* send(parentThreadId, "first", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, "child task"); + assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(childThreadId, 2)))); + const grandchildThreadId = yield* delegate(childThreadId, "grandchild task"); + yield* send(childThreadId, "child follow-up", "queue_after_active"); + + const now = yield* DateTime.now; + const firstRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; + yield* projections.apply({ + id: EventId.make("event:stop-parent:first-run-completed"), + type: "run.updated", + threadId: parentThreadId, + runId: firstRun.id, + occurredAt: now, + payload: { ...firstRun, status: "completed", completedAt: now }, + }); + yield* send(parentThreadId, "second", "start_immediately"); + const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!; + + const stopCommandId = CommandId.make("stop-parent"); + yield* orchestrator.dispatch( + type === "thread.stop" + ? { type, commandId: stopCommandId, threadId: parentThreadId } + : { + type, + commandId: stopCommandId, + threadId: parentThreadId, + runId: secondRun.id, + holdQueue: true, + }, + ); + + assert.deepEqual(yield* threadState(parentThreadId), { + runs: ["completed", "interrupted"], + watched: [], + }); + const parent = yield* orchestrator.getThreadProjection(parentThreadId); + assert.equal(parent.subagents[0]?.completionDelivery?.state, "disposed"); + const effects = yield* sql<{ readonly effect_type: string }>` + SELECT effect_type FROM orchestration_v2_effect_outbox WHERE command_id = ${stopCommandId} + `; + assert.include( + effects.map((row) => row.effect_type), + "delegated-tasks.stop", + ); + + yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); + assert.deepEqual(yield* threadState(childThreadId), { + runs: ["interrupted", "queued:held"], + watched: [], + }); + assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]); + + yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); + assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]); + // The walk stops each level itself, so the child stops it sends queue no walk of their own. + const walks = yield* sql<{ readonly command_id: string }>` + SELECT command_id FROM orchestration_v2_effect_outbox + WHERE effect_type = 'delegated-tasks.stop' + `; + assert.deepEqual( + walks.map((row) => row.command_id), + [stopCommandId], + ); + }).pipe(Effect.provide(layerTest)), -it.effect("Stop ends watches, holds queues, and stops the delegated tasks under the thread", () => +); + +it.effect("a repeated Stop that changes nothing still stops the delegated tasks", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; - const threads = yield* ThreadManagementService.ThreadManagementService; const sql = yield* SqlClient.SqlClient; + const parentThreadId = ThreadId.make("thread:stop-again"); - const parentThreadId = ThreadId.make("thread:stop-parent"); yield* createWatchingThread(parentThreadId, 1); yield* send(parentThreadId, "first", "start_immediately"); - const childThreadId = yield* delegate(parentThreadId, "child task"); + yield* delegate(parentThreadId, "stop-again child"); - assert.isTrue(Exit.isFailure(yield* Effect.exit(watch(childThreadId, 2)))); - const grandchildThreadId = yield* delegate(childThreadId, "grandchild task"); - yield* send(childThreadId, "child follow-up", "queue_after_active"); - const now = yield* DateTime.now; - const firstRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; + const run = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!; yield* projections.apply({ + id: EventId.make("event:stop-again:run-completed"), - id: EventId.make("event:stop-parent:first-run-completed"), type: "run.updated", threadId: parentThreadId, + runId: run.id, - runId: firstRun.id, occurredAt: now, - payload: { ...firstRun, status: "completed", completedAt: now }, + payload: { ...run, status: "completed", completedAt: now }, }); - yield* send(parentThreadId, "second", "start_immediately"); - const secondRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!; - - const stopCommandId = CommandId.make("stop-parent"); yield* orchestrator.dispatch({ - type: "run.interrupt", - commandId: stopCommandId, + type: "thread.stop", + commandId: CommandId.make("stop-again:first"), threadId: parentThreadId, - runId: secondRun.id, - holdQueue: true, }); + // The first Stop already disposed the delivery and ended the watch, so this one writes no + // events, like Stop on a thread an older server stopped without stopping its children. + const again = yield* orchestrator.dispatch({ + type: "thread.stop", - assert.deepEqual(yield* threadState(parentThreadId), { - runs: ["completed", "interrupted"], + commandId: CommandId.make("stop-again:second"), + threadId: parentThreadId, - watched: [], }); - const parent = yield* orchestrator.getThreadProjection(parentThreadId); - assert.equal(parent.subagents[0]?.completionDelivery?.state, "disposed"); - const effects = yield* sql<{ readonly effect_type: string }>` - SELECT effect_type FROM orchestration_v2_effect_outbox WHERE command_id = ${stopCommandId} + assert.lengthOf(again.storedEvents, 0); + const walks = yield* sql<{ readonly command_id: string }>` + SELECT command_id FROM orchestration_v2_effect_outbox + WHERE effect_type = 'delegated-tasks.stop' `; + assert.deepEqual( + walks.map((row) => row.command_id), - assert.include( + ["stop-again:first", "stop-again:second"], - effects.map((row) => row.effect_type), - "delegated-tasks.stop", ); - - yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); - assert.deepEqual(yield* threadState(childThreadId), { - runs: ["interrupted", "queued:held"], - watched: [], - }); - assert.deepEqual((yield* threadState(grandchildThreadId)).runs, ["interrupted"]); - - yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); - assert.deepEqual((yield* threadState(childThreadId)).runs, ["interrupted", "queued:held"]); }).pipe(Effect.provide(layerTest)), ); ``` ## packages/contracts/src/orchestrationV2.ts Status: modified; +5; −0 ```diff @@ -2577,6 +2577,11 @@ commandId: CommandId, threadId: ThreadId, reason: Schema.optional(Schema.String), + /** + * False when the caller stops the thread's delegated tasks itself, as the server's tree walks + * do under their caller's dispatch limit. Otherwise Stop also stops every task under the thread. + */ + stopDelegatedTasks: Schema.optional(Schema.Boolean), }), Schema.Struct({ type: Schema.Literal("thread.create"), ```