From 59366120cfc7018476a8519237a9128e1589e597 Mon Sep 17 00:00:00 2001 From: syberia Date: Mon, 19 Oct 2026 11:16:40 +0200 Subject: [PATCH] fix(server): pressing Stop now clears orphaned delegated follow-ups --- .../src/orchestration-v2/Orchestrator.ts | 243 ++++++++++-------- .../src/orchestration-v2/ThreadStop.test.ts | 98 +++++-- 2 files changed, 217 insertions(+), 124 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 4464d09916e33..048ab45e29026 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9594,6 +9594,121 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio return undefined; }); + const cancelIdleOrphans = (input: { + readonly command: Extract< + OrchestrationV2ServerCommand, + { readonly type: "run.interrupt" | "thread.stop" } + >; + readonly events: Ref.Ref>; + readonly projection: Pick; + }) => + Effect.gen(function* () { + const { command, events, projection } = input; + const pending = yield* Ref.get(events); + const children = new Map< + ThreadId, + ProjectionRecords<"runs" | "providerThreads" | "runtimeRequests" | "turnItems"> | null + >(); + const idleChildren = new Map(); + let nodes: ReadonlyArray | undefined; + for (const recorded of projection.subagents) { + const updated = pending.findLast( + (event) => event.type === "subagent.updated" && event.payload.id === recorded.id, + ); + const task = updated?.type === "subagent.updated" ? updated.payload : recorded; + if ( + task.origin !== "app_owned" || + task.childThreadId === null || + !isOrchestrationV2WorkActive(task.status) + ) + continue; + const childThreadId = task.childThreadId; + const child = children.has(childThreadId) + ? (children.get(childThreadId) ?? null) + : yield* projectionStore + .getThreadRecords( + childThreadId, + ["runs", "providerThreads", "runtimeRequests", "turnItems"], + { + turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], + turnItemStatuses: ["pending", "running", "waiting"], + }, + ) + .pipe( + Effect.catchTags({ + ProjectionStoreThreadNotFoundError: () => Effect.succeed(null), + }), + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: childThreadId, cause }), + ), + ); + children.set(childThreadId, child); + if (child === null) continue; + if ( + child.thread.lineage.parentThreadId !== command.threadId || + child.thread.lineage.relationshipToParent !== "subagent" || + (child.thread.forkedFrom?.type === "node" && + child.thread.forkedFrom.nodeId === task.id) || + child.runs.some((run) => run.delegatedTaskId === task.id) + ) + continue; + const idle = + idleChildren.get(task.childThreadId) ?? + (!child.runs.some((run) => run.status === "queued" || isBlockingRun(run)) && + !child.providerThreads.some( + (thread) => + thread.status === "active" || (thread.pendingBackgroundTasks?.length ?? 0) > 0, + ) && + !child.runtimeRequests.some((request) => request.status === "pending") && + child.turnItems.length === 0 && + Option.isNone(yield* Effect.findFirst(child.runs, awaitsRestartContinuation))); + idleChildren.set(task.childThreadId, idle); + if (!idle) continue; + // Stop can cancel an idle orphan without guessing which completed run owned its result. + const now = yield* DateTime.now; + const emitEvent = emit(events, command); + yield* emitEvent({ + type: "subagent.updated", + threadId: command.threadId, + nodeId: task.id, + occurredAt: now, + payload: { + ...task, + status: "cancelled", + completedAt: now, + updatedAt: now, + completionDelivery: { state: "disposed", observedByRunId: null }, + }, + }); + nodes ??= (yield* projectionStore.getThreadRecords(command.threadId, ["nodes"])).nodes; + const node = nodes.find((candidate) => candidate.id === task.id); + if (node !== undefined) + yield* emitEvent({ + type: "node.updated", + threadId: command.threadId, + nodeId: node.id, + occurredAt: now, + payload: { ...node, status: "cancelled", completedAt: now }, + }); + for (const item of projection.turnItems) { + if (item.type !== "subagent" || item.subagentId !== task.id) continue; + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + nodeId: task.id, + occurredAt: now, + payload: { ...item, status: "cancelled", completedAt: now, updatedAt: now }, + }); + } + } + }).pipe( + Effect.mapError((cause) => + Schema.is(OrchestratorProjectionError)(cause) + ? cause + : new OrchestratorProjectionError({ threadId: input.command.threadId, cause }), + ), + ); + const dispatchThreadStop = ( command: Extract, events: Ref.Ref>, @@ -9611,111 +9726,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); if (projection.thread.deletedAt !== null) return undefined; - const cancelIdleOrphans = Effect.gen(function* () { - const pending = yield* Ref.get(events); - const children = new Map< - ThreadId, - ProjectionRecords<"runs" | "providerThreads" | "runtimeRequests" | "turnItems"> | null - >(); - const idleChildren = new Map(); - let nodes: ReadonlyArray | undefined; - for (const recorded of projection.subagents) { - const updated = pending.findLast( - (event) => event.type === "subagent.updated" && event.payload.id === recorded.id, - ); - const task = updated?.type === "subagent.updated" ? updated.payload : recorded; - if ( - task.origin !== "app_owned" || - task.childThreadId === null || - !isOrchestrationV2WorkActive(task.status) - ) - continue; - const childThreadId = task.childThreadId; - const child = children.has(childThreadId) - ? (children.get(childThreadId) ?? null) - : yield* projectionStore - .getThreadRecords( - childThreadId, - ["runs", "providerThreads", "runtimeRequests", "turnItems"], - { - turnItemTypes: ["command_execution", "dynamic_tool", "subagent"], - turnItemStatuses: ["pending", "running", "waiting"], - }, - ) - .pipe( - Effect.catchTags({ - ProjectionStoreThreadNotFoundError: () => Effect.succeed(null), - }), - Effect.mapError( - (cause) => new OrchestratorProjectionError({ threadId: childThreadId, cause }), - ), - ); - children.set(childThreadId, child); - if (child === null) continue; - if ( - child.thread.lineage.parentThreadId !== command.threadId || - child.thread.lineage.relationshipToParent !== "subagent" || - (child.thread.forkedFrom?.type === "node" && - child.thread.forkedFrom.nodeId === task.id) || - child.runs.some((run) => run.delegatedTaskId === task.id) - ) - continue; - const idle = - idleChildren.get(task.childThreadId) ?? - (!child.runs.some((run) => run.status === "queued" || isBlockingRun(run)) && - !child.providerThreads.some( - (thread) => - thread.status === "active" || (thread.pendingBackgroundTasks?.length ?? 0) > 0, - ) && - !child.runtimeRequests.some((request) => request.status === "pending") && - child.turnItems.length === 0 && - Option.isNone(yield* Effect.findFirst(child.runs, awaitsRestartContinuation))); - idleChildren.set(task.childThreadId, idle); - if (!idle) continue; - // Stop can cancel an idle orphan without guessing which completed run owned its result. - const now = yield* DateTime.now; - const emitEvent = emit(events, command); - yield* emitEvent({ - type: "subagent.updated", - threadId: command.threadId, - nodeId: task.id, - occurredAt: now, - payload: { - ...task, - status: "cancelled", - completedAt: now, - updatedAt: now, - completionDelivery: { state: "disposed", observedByRunId: null }, - }, - }); - nodes ??= (yield* projectionStore.getThreadRecords(command.threadId, ["nodes"])).nodes; - const node = nodes.find((candidate) => candidate.id === task.id); - if (node !== undefined) - yield* emitEvent({ - type: "node.updated", - threadId: command.threadId, - nodeId: node.id, - occurredAt: now, - payload: { ...node, status: "cancelled", completedAt: now }, - }); - for (const item of projection.turnItems) { - if (item.type !== "subagent" || item.subagentId !== task.id) continue; - yield* emitEvent({ - type: "turn-item.updated", - threadId: command.threadId, - nodeId: task.id, - occurredAt: now, - payload: { ...item, status: "cancelled", completedAt: now, updatedAt: now }, - }); - } - } - }).pipe( - Effect.mapError((cause) => - Schema.is(OrchestratorProjectionError)(cause) - ? cause - : new OrchestratorProjectionError({ threadId: command.threadId, cause }), - ), - ); const latestRun = projection.runs.findLast((run) => run.status !== "queued"); const target = projection.runs.findLast( @@ -9754,7 +9764,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const enqueued = yield* Ref.get(interruptEffects); yield* Ref.update(events, (existing) => [...existing, ...written]); yield* Ref.update(effects, (existing) => [...existing, ...enqueued]); - yield* cancelIdleOrphans; + yield* cancelIdleOrphans({ command, events, projection }); return interrupted.value; } yield* Effect.logWarning("thread.stop could not interrupt the running turn", { @@ -9775,7 +9785,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* markStoppedRun({ command, events, thread: projection.thread, run, now }); } yield* holdStoppedThread({ command, events, projection, cohortRunIds: [], now }); - yield* cancelIdleOrphans; + yield* cancelIdleOrphans({ command, events, projection }); return undefined; }); @@ -11100,6 +11110,25 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "run.interrupt": cancelUnsettledEffects = yield* dispatchRunInterrupt(command, events, effects); if (command.holdQueue === true) { + // Stop must still commit when a child can't be read, so a failed repair's + // partial events are dropped. + const repairEvents = yield* Ref.make(yield* Ref.get(events)); + const repaired = yield* Effect.exit( + loadProjectionForCommand(command, ["subagents", "turnItems"], { + turnItemTypes: ["subagent"], + turnItemStatuses: ["pending", "running", "waiting"], + }).pipe( + Effect.flatMap((projection) => + cancelIdleOrphans({ command, events: repairEvents, projection }), + ), + ), + ); + if (Exit.isSuccess(repaired)) yield* Ref.set(events, yield* Ref.get(repairEvents)); + else + yield* Effect.logWarning("Stop could not cancel orphaned delegated follow-ups", { + threadId: command.threadId, + cause: repaired.cause, + }); yield* Ref.update(effects, (existing) => [ ...existing, { diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 12db927d9c6d0..d428b826f0739 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -161,21 +161,22 @@ const threadState = (threadId: ThreadId) => }); it.effect.each([ - "idle", - "archived", - "missing", - "running", - "queued", - "waiting", - "correlated", - "provider-active", - "roster", - "background", - "question", - "restart", + ["idle", "thread.stop"], + ["idle", "run.interrupt"], + ["archived", "thread.stop"], + ["missing", "thread.stop"], + ["running", "thread.stop"], + ["queued", "thread.stop"], + ["waiting", "thread.stop"], + ["correlated", "thread.stop"], + ["provider-active", "thread.stop"], + ["roster", "thread.stop"], + ["background", "thread.stop"], + ["question", "thread.stop"], + ["restart", "thread.stop"], ] as const)( - "Stop clears an orphaned follow-up only when its child is finished (%s)", - (childState) => + "Stop clears an orphaned follow-up only when its child is finished (%s via %s)", + ([childState, type]) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; @@ -367,6 +368,41 @@ it.effect.each([ occurredAt: now, payload: { ...parent.runs[0]!, status: "completed", completedAt: now }, }); + if (type === "run.interrupt") { + const run = parent.runs[0]!; + yield* apply({ + type: "provider-turn.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { + id: ProviderTurnId.make("provider-turn:orphan-parent"), + providerThreadId: run.providerThreadId!, + nodeId: run.rootNodeId!, + runAttemptId: run.activeAttemptId!, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }); + yield* apply({ + type: "run-attempt.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { ...parent.attempts[0]!, status: "completed", completedAt: now }, + }); + yield* apply({ + type: "node.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { + ...parent.nodes.find((node) => node.id === run.rootNodeId)!, + status: "completed", + completedAt: now, + }, + }); + } if (childState === "archived") { yield* apply({ type: "thread.archived", @@ -382,11 +418,15 @@ it.effect.each([ ), 1, ); - yield* orchestrator.dispatch({ - type: "thread.stop", + const stop = { commandId: CommandId.make(`stop-orphan:${childState}`), threadId: parentThreadId, - }); + }; + yield* orchestrator.dispatch( + type === "run.interrupt" + ? { ...stop, type, runId: parent.runs.at(-1)!.id, holdQueue: true } + : { ...stop, type }, + ); const after = yield* orchestrator.getThreadProjection(parentThreadId); assert.isEmpty((after.thread.pullRequests ?? []).filter((link) => link.watch !== undefined)); const expected = childState === "idle" || childState === "archived" ? "cancelled" : "running"; @@ -421,6 +461,30 @@ it.effect.each([ }).pipe(Effect.provide(layerTest)), ); +it.effect("Stop still stops the thread when a delegated child cannot be read", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sql = yield* SqlClient.SqlClient; + const parentThreadId = ThreadId.make("thread:unreadable-child-parent"); + yield* createWatchingThread(parentThreadId, 1); + yield* send(parentThreadId, "parent", "start_immediately"); + const childThreadId = yield* delegate(parentThreadId, "unreadable child"); + yield* sql` + UPDATE orchestration_v2_projection_runs + SET payload_json = '{"obsolete":true}' + WHERE thread_id = ${childThreadId} + `; + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make("stop:unreadable-child"), + threadId: parentThreadId, + runId: (yield* orchestrator.getThreadProjection(parentThreadId)).runs.at(-1)!.id, + holdQueue: true, + }); + assert.deepEqual(yield* threadState(parentThreadId), { runs: ["interrupted"], watched: [] }); + }).pipe(Effect.provide(layerTest)), +); + it.effect("Stop ends watches, holds queues, and stops the delegated tasks under the thread", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;