fix(server): keep retried linked Stops within their original work

apps/server/src/orchestration-v2/EventSink.ts

@@ -189,6 +189,11 @@ export interface EventSinkV2Shape {
   readonly latestSequence: (input?: {
     readonly threadId?: ThreadId;
   }) => Effect.Effect<number, EventSinkV2Error>;
+  readonly read: (input: {
+    readonly threadId: ThreadId;
+    readonly throughSequence: number;
+    readonly eventType: OrchestrationV2DomainEvent["type"];
+  }) => Stream.Stream<OrchestrationV2StoredEvent, EventSinkStreamError>;
   readonly readByCommandId: (input: {
     readonly commandId: CommandId;
   }) => Stream.Stream<OrchestrationV2StoredEvent, EventSinkV2Error>;
@@ -651,12 +656,10 @@ const layerBase: Layer.Layer<
             commandId: input.commandId,
             events: normalized,
           });
-          // A Stop that changes nothing can still queue the stop of its delegated tasks.
+          // An eventless Stop still needs a current cutoff for its child traversal.
           const sequence =
             storedEvents.at(-1)?.sequence ??
-            (input.effects.length === 0
-              ? undefined
-              : yield* eventStore.latestSequence({ threadId: input.threadId }));
+            (input.effects.length === 0 ? undefined : yield* eventStore.latestSequence());
           if (sequence === undefined) {
             return yield* Effect.die(
               new Error(`Command ${input.commandId} produced no orchestration events.`),
@@ -977,6 +980,14 @@ const layerBase: Layer.Layer<
               }),
           ),
         ),
+      read: (input) =>
+        eventStore
+          .read(input)
+          .pipe(
+            Stream.mapError(
+              (cause) => new EventSinkStreamError({ threadId: input.threadId, cause }),
+            ),
+          ),
       readByCommandId: (input) =>
         eventStore.readByCommandId(input).pipe(
           Stream.mapError(

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -411,6 +411,12 @@ export interface OrchestratorV2Shape {
   readonly getThreadEventSequence: (
     threadId: ThreadId,
   ) => Effect.Effect<number, OrchestratorV2Error>;
+  /** Committed history at a fixed sequence, without subscribing to live events. */
+  readonly readStoredEvents: (input: {
+    readonly threadId: ThreadId;
+    readonly throughSequence: number;
+    readonly eventType: OrchestrationV2DomainEvent["type"];
+  }) => Stream.Stream<OrchestrationV2StoredEvent, OrchestratorV2Error>;
   readonly streamStoredEvents: Stream.Stream<OrchestrationV2StoredEvent, OrchestratorV2Error>;
   readonly streamStoredEventsFrom: (input?: {
     readonly threadId?: ThreadId;
@@ -10026,7 +10032,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         ? undefined
         : projection.providerThreads.find((candidate) => candidate.id === run?.providerThreadId);
     const hasBackgroundWork =
-      run?.id === projection.runs.findLast((candidate) => candidate.status !== "queued")?.id &&
+      run?.id === latestExecutedRun(projection.runs)?.id &&
       derivePendingBackgroundWork({
         latestRun: run,
         providerThreads: projection.providerThreads,
@@ -10562,7 +10568,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
   });
 
   const dispatchThreadStop = Effect.fn("Orchestrator.dispatchThreadStop")(function* (
-    command: Extract<OrchestrationV2ServerCommand, { readonly type: "thread.stop" }>,
+    command: Extract<OrchestrationV2InternalCommand, { readonly type: "thread.stop" }>,
     events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
     effects: Ref.Ref<Array<PendingOrchestrationEffectV2>>,
   ) {
@@ -10585,7 +10591,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       )
         ? Effect.void
         : Ref.update(effects, (existing) => [...existing, delegatedTasksStopEffect(command)]);
-    const latestRun = projection.runs.findLast((run) => run.status !== "queued");
+    const executed = latestExecutedRun(projection.runs);
+    const latestRun = executed ?? undefined;
     const target =
       projection.runs.findLast(
         (run) =>
@@ -10600,6 +10607,37 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       }).length > 0
         ? latestRun
         : undefined);
+    const reachedRun = target ?? executed;
+    if (command.throughSequence !== undefined && reachedRun !== null) {
+      // A failed child Stop cannot reach work started after the original parent Stop.
+      const hadStarted = yield* eventSink
+        .read({
+          threadId: command.threadId,
+          throughSequence: command.throughSequence,
+          eventType: "run.created",
+        })
+        .pipe(
+          Stream.concat(
+            eventSink.read({
+              threadId: command.threadId,
+              throughSequence: command.throughSequence,
+              eventType: "run.updated",
+            }),
+          ),
+          Stream.runFold(
+            () => false,
+            (found, stored) =>
+              found ||
+              ((stored.event.type === "run.created" || stored.event.type === "run.updated") &&
+                stored.event.runId === reachedRun.id &&
+                stored.event.payload.status !== "queued"),
+          ),
+          Effect.mapError(
+            (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }),
+          ),
+        );
+      if (!hadStarted) return undefined;
+    }
     if (target !== undefined) {
       const interruptEvents = yield* Ref.make<Array<OrchestrationV2DomainEvent>>([]);
       const interruptEffects = yield* Ref.make<Array<PendingOrchestrationEffectV2>>([]);
@@ -10636,7 +10674,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
     const now = yield* DateTime.now;
     // Record that Stop reached the turn it could not interrupt and the thread's last executed
     // run, so a late tool call from their agent starts nothing.
-    const executed = latestExecutedRun(projection.runs);
     const reached = new Map<RunId, OrchestrationV2Run>();
     if (target !== undefined) reached.set(target.id, target);
     if (executed !== null && !hasLiveRun({ runs: [executed] })) reached.set(executed.id, executed);
@@ -12732,6 +12769,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
         .latestSequence({ threadId })
         .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause })));
     }),
+    readStoredEvents: (input) =>
+      eventSink
+        .read(input)
+        .pipe(Stream.mapError((cause) => new OrchestratorDomainEventStreamError({ cause }))),
     streamStoredEvents: eventSink.stream().pipe(
       Stream.mapError(
         (cause) =>

apps/server/src/orchestration-v2/ThreadManagementService.ts

@@ -44,6 +44,7 @@ import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Ref from "effect/Ref";
+import * as Result from "effect/Result";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 
@@ -1162,38 +1163,83 @@ const make = Effect.gen(function* () {
     return { matched, unmatched: input.taskIds.filter((taskId) => !matchedIds.has(taskId)) };
   });
 
+  const stopDelegatedTasksThrough = Effect.fn("ThreadManagementService.stopDelegatedTasksThrough")(
+    function* (
+      input: Parameters<ThreadManagementServiceShape["stopDelegatedTasks"]>[0],
+      throughSequence: number | undefined,
+    ): Effect.fn.Return<
+      void,
+      Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
+    > {
+      // A retry must not stop tasks delegated after the original Stop, including
+      // when an earlier walk stopped some children and failed on another.
+      // Follow-up tasks reuse a child thread, so the set walks its subtree once.
+      const childThreadIds = new Set(
+        throughSequence === undefined
+          ? (yield* orchestrator.getThreadRecords(input.threadId, ["subagents"])).subagents.flatMap(
+              (task) =>
+                task.origin === "app_owned" && task.childThreadId !== null
+                  ? [task.childThreadId]
+                  : [],
+            )
+          : yield* orchestrator
+              .readStoredEvents({
+                threadId: input.threadId,
+                throughSequence,
+                eventType: "subagent.updated",
+              })
+              .pipe(
+                Stream.filterMap((stored) =>
+                  stored.event.type === "subagent.updated" &&
+                  stored.event.payload.origin === "app_owned" &&
+                  stored.event.payload.childThreadId !== null
+                    ? Result.succeed(stored.event.payload.childThreadId)
+                    : Result.failVoid,
+                ),
+                Stream.runCollect,
+              ),
+      );
+      const failures: Array<
+        Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
+      > = [];
+      for (const threadId of childThreadIds) {
+        yield* dispatch({
+          type: "thread.stop",
+          commandId: CommandId.make(`${input.commandId}:stop:${threadId}`),
+          threadId,
+          stopDelegatedTasks: false,
+          ...(throughSequence === undefined ? {} : { throughSequence }),
+          ...(input.reason === undefined ? {} : { reason: input.reason }),
+        }).pipe(
+          Effect.andThen(stopDelegatedTasksThrough({ ...input, threadId }, throughSequence)),
+          Effect.catch((error) =>
+            Effect.logWarning("Unable to stop a delegated task", {
+              parentThreadId: input.threadId,
+              threadId,
+              error,
+            }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))),
+          ),
+        );
+      }
+      if (failures[0] !== undefined) return yield* failures[0];
+    },
+  );
+
   const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = Effect.fn(
     "ThreadManagementService.stopDelegatedTasks",
   )(function* (input) {
-    const { subagents } = yield* orchestrator.getThreadRecords(input.threadId, ["subagents"]);
-    const failures: Array<
-      Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError
-    > = [];
-    const stoppedChildThreadIds = new Set<ThreadId>();
-    for (const task of subagents) {
-      if (task.origin !== "app_owned" || task.childThreadId === null) continue;
-      const threadId = task.childThreadId;
-      // Follow-up tasks reuse a child thread; walk its subtree only once.
-      if (stoppedChildThreadIds.has(threadId)) continue;
-      stoppedChildThreadIds.add(threadId);
-      yield* dispatch({
-        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 })),
-        Effect.catch((error) =>
-          Effect.logWarning("Unable to stop a delegated task", {
-            parentThreadId: input.threadId,
-            threadId,
-            error,
-          }).pipe(Effect.andThen(Effect.sync(() => failures.push(error)))),
+    const receipt = yield* receipts
+      .getByCommandId(input.commandId)
+      .pipe(
+        Effect.mapError(
+          (cause) =>
+            new Orchestrator.OrchestratorProjectionError({ threadId: input.threadId, cause }),
         ),
       );
-    }
-    if (failures[0] !== undefined) return yield* failures[0];
+    return yield* stopDelegatedTasksThrough(
+      input,
+      Option.isSome(receipt) ? receipt.value.resultSequence : undefined,
+    );
   });
 
   return ThreadManagementService.of({