fix(server): cascade parent lifecycle to subagent descendants

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

@@ -31,9 +31,19 @@ export class CommandReceiptStoreReadError extends Schema.TaggedError<CommandRece
   }
 }
 
+class CommandReceiptStoreLifecycleError extends Schema.TaggedError<CommandReceiptStoreLifecycleError>()(
+  "CommandReceiptStoreLifecycleError",
+  { threadId: ThreadId, cause: Schema.Defect() },
+) {
+  override get message(): string {
+    return `Failed to read orchestration V2 lifecycle receipt for thread ${this.threadId}.`;
+  }
+}
+
 export const CommandReceiptStoreV2Error = Schema.Union([
   CommandReceiptStoreWriteError,
   CommandReceiptStoreReadError,
+  CommandReceiptStoreLifecycleError,
 ]);
 export type CommandReceiptStoreV2Error = typeof CommandReceiptStoreV2Error.Type;
 
@@ -65,6 +75,9 @@ export type ProjectCommandReceiptV2 = typeof ProjectCommandReceiptV2.Type;
 type AnyCommandReceiptV2 = CommandReceiptV2 | ProjectCommandReceiptV2;
 
 export interface CommandReceiptStoreV2Shape {
+  readonly getLatestLifecycle: (
+    threadId: ThreadId,
+  ) => Effect.Effect<CommandReceiptV2 | null, CommandReceiptStoreV2Error>;
   readonly insertIfAbsent: (
     receipt: AnyCommandReceiptV2,
   ) => Effect.Effect<boolean, CommandReceiptStoreV2Error>;
@@ -137,6 +150,13 @@ const layerBase: Layer.Layer<
     const receipts = yield* OrchestrationCommandReceipts.OrchestrationCommandReceiptRepository;
 
     return CommandReceiptStoreV2.of({
+      getLatestLifecycle: (threadId) =>
+        receipts.getLatestLifecycle({ threadId }).pipe(
+          Effect.flatMap((receipt) =>
+            Option.isSome(receipt) ? fromApplicationReceipt(receipt.value) : Effect.succeed(null),
+          ),
+          Effect.mapError((cause) => new CommandReceiptStoreLifecycleError({ threadId, cause })),
+        ),
       insertIfAbsent: (receipt) =>
         receipts.insertIfAbsent(toApplicationReceipt(receipt)).pipe(
           Effect.mapError(
@@ -200,3 +220,52 @@ const layerBase: Layer.Layer<
 export const layer = layerBase.pipe(Layer.provide(OrchestrationCommandReceipts.layer));
 
 export const layerFromApplicationReceipts = layerBase;
+
+/** Rejections are permanent, so a cascade retry takes the first unused or accepted suffix. */
+export const cascadeCommandId = Effect.fn("CommandReceiptStore.cascadeCommandId")(function* (
+  receipts: CommandReceiptStoreV2Shape,
+  base: CommandId,
+) {
+  for (let retry = 0; retry < 32; retry++) {
+    const id = retry === 0 ? base : CommandId.make(`${base}:retry:${retry}`);
+    const receipt = yield* receipts.getByCommandId(id);
+    if (Option.isNone(receipt) || receipt.value.status === "accepted") return id;
+  }
+  return undefined;
+});
+
+/** A lifecycle choice made after the cascade's source command, other than `ownCommandId`. */
+export const supersedesCascade = (
+  latest: CommandReceiptV2 | null,
+  source: CommandReceiptV2,
+  ownCommandId: CommandId,
+) =>
+  latest !== null &&
+  latest.resultSequence > source.resultSequence &&
+  latest.commandId !== ownCommandId;
+
+/** Whether a cascade from an archive of `rootThreadId`, along `path`, archived this thread. */
+export const archivedByCascade = Effect.fn("CommandReceiptStore.archivedByCascade")(function* (
+  receipts: CommandReceiptStoreV2Shape,
+  latest: CommandReceiptV2 | null,
+  rootThreadId: ThreadId,
+  path: ReadonlyArray<ThreadId>,
+) {
+  if (latest?.commandType !== "thread.archive") return false;
+  let rootCommandId: string = latest.commandId;
+  for (const threadId of path.toReversed()) {
+    const segment = `:subagent:${threadId}`;
+    if (!rootCommandId.endsWith(segment)) rootCommandId = rootCommandId.replace(/:retry:\d+$/, "");
+    if (!rootCommandId.endsWith(segment)) return false;
+    rootCommandId = rootCommandId.slice(0, -segment.length);
+  }
+  const root = Option.getOrNull(yield* receipts.getByCommandId(CommandId.make(rootCommandId)));
+  const valid = root?.status === "accepted" && root.threadId === rootThreadId;
+  // Derive the IDs again, so a user ID only shaped like a cascade retry is not counted.
+  let derived = valid && root.commandType === "thread.archive" ? root.commandId : undefined;
+  for (const threadId of path) {
+    if (derived === undefined) return false;
+    derived = yield* cascadeCommandId(receipts, CommandId.make(`${derived}:subagent:${threadId}`));
+  }
+  return derived === latest.commandId && (root?.resultSequence ?? 0) <= latest.resultSequence;
+});

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

@@ -12,7 +12,9 @@ import {
   ProviderUserInputAnswers,
   ProviderThreadId,
   ProviderTurnId,
+  ProviderInteractionMode,
   RunId,
+  RuntimeMode,
   RuntimeRequestId,
   ThreadId,
 } from "@t3tools/contracts";
@@ -30,6 +32,13 @@ import * as Schema from "effect/Schema";
 import * as SqlClient from "effect/sql/SqlClient";
 
 import { forkParked } from "../serverActivation.ts";
+import type { CommandReceiptV2 } from "./CommandReceiptStore.ts";
+
+/** Set while a cascade dispatches a descendant's lifecycle command; delete needs no source. */
+export const SubagentCascade = Context.Reference<
+  | { readonly source: CommandReceiptV2 | undefined; readonly path: ReadonlyArray<ThreadId> }
+  | undefined
+>("t3/orchestration-v2/SubagentCascade", { defaultValue: () => undefined });
 
 export const OrchestrationEffectRequestV2 = Schema.Union([
   Schema.Struct({
@@ -118,6 +127,13 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
     type: Schema.Literal("attachment.cleanup"),
     attachmentIds: Schema.Array(Schema.String),
   }),
+  Schema.Struct({
+    type: Schema.Literal("subagent-threads.cascade"),
+    operation: Schema.Literals(["thread.archive", "thread.unarchive", "thread.delete"]),
+    limit: Schema.optional(
+      Schema.Struct({ runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode }),
+    ),
+  }),
   Schema.Struct({
     type: Schema.Literal("thread-title.generate"),
     kind: Schema.Union([
@@ -141,6 +157,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
   "terminal.cleanup",
   "preview.cleanup",
   "attachment.cleanup",
+  "subagent-threads.cascade",
   "thread-title.generate",
   "delegated-tasks.stop",
 ] as const satisfies ReadonlyArray<OrchestrationEffectRequestV2["type"]>;

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

@@ -29,6 +29,7 @@ import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
 import * as RuntimeRequestService from "./RuntimeRequestService.ts";
 import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
 import * as ThreadManagementService from "./ThreadManagementService.ts";
+import { DispatchModeLimit } from "./DispatchModeLimit.ts";
 import * as ServerSettings from "../serverSettings.ts";
 import { continueRestartedRun } from "./RestartContinuation.ts";
 
@@ -595,6 +596,24 @@ export const layerExecutor: Layer.Layer<
                   }),
               ),
             );
+          case "subagent-threads.cascade":
+            return threads
+              .cascadeSubagentLifecycle({
+                commandId: effect.commandId,
+                threadId: effect.threadId,
+                operation: effect.request.operation,
+              })
+              .pipe(
+                Effect.provideService(DispatchModeLimit, effect.request.limit),
+                Effect.mapError(
+                  (cause) =>
+                    new OrchestrationEffectExecutionError({
+                      effectId: effect.id,
+                      effectType: effect.request.type,
+                      cause,
+                    }),
+                ),
+              );
           case "thread-title.generate":
             return threadTitleRegeneration
               .execute({

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

@@ -116,7 +116,12 @@ import {
   isNativeMaintenanceCommand,
   resolveMessageDispatchIntent,
 } from "./CommandPolicy.ts";
-import { CommandReceiptStoreV2, type CommandReceiptV2 } from "./CommandReceiptStore.ts";
+import {
+  archivedByCascade,
+  CommandReceiptStoreV2,
+  type CommandReceiptV2,
+  supersedesCascade,
+} from "./CommandReceiptStore.ts";
 import {
   ContextHandoffServiceV2,
   readForkHistoryItems,
@@ -460,6 +465,9 @@ export interface OrchestratorV2Shape {
   readonly getThreadProjection: (
     threadId: ThreadId,
   ) => Effect.Effect<OrchestrationV2ThreadProjection, OrchestratorV2Error>;
+  readonly getSubagentChildThreads: (
+    ...args: Parameters<ProjectionStoreV2["Service"]["getSubagentChildThreads"]>
+  ) => Effect.Effect<ReadonlyArray<OrchestrationV2AppThread>, OrchestratorV2Error>;
   readonly getCheckpointContext: (
     threadId: ThreadId,
   ) => Effect.Effect<ProjectionCheckpointContext, OrchestratorV2Error>;
@@ -1785,6 +1793,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause })),
     );
 
+  // A cascade walks the whole subtree itself, so its own commands start none.
+  const subagentCascadeEffects = Effect.fn("Orchestrator.subagentCascadeEffects")(function* (
+    { commandId, threadId }: { readonly commandId: CommandId; readonly threadId: ThreadId },
+    operation: "thread.archive" | "thread.unarchive" | "thread.delete",
+  ) {
+    if (yield* EffectOutbox.SubagentCascade) return [];
+    if ((yield* projectionStore.getSubagentChildThreads(threadId)).length === 0) return [];
+    // The outbox schema persists only the caller's runtime and interaction modes.
+    const limit = yield* DispatchModeLimit;
+    const request = { type: "subagent-threads.cascade", operation, limit } as const;
+    const id = `effect:${commandId}:subagent-threads.cascade`;
+    return [{ id, commandId, threadId, request } satisfies PendingOrchestrationEffectV2];
+  });
+
   /** Side chats live inside their parent, so deleting the parent deletes them too. */
   const liveSideChatIds = Effect.fn("Orchestrator.liveSideChatIds")(
     function* (threadId: ThreadId) {
@@ -5176,6 +5198,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       }
     }
 
+    if (command.type === "thread.archive" || command.type === "thread.unarchive") {
+      const cascade = yield* subagentCascadeEffects(command, command.type).pipe(
+        mapDispatchError(command),
+      );
+      yield* Ref.update(effects, (existing) => [...existing, ...cascade]);
+    }
+
     if (command.type === "thread.metadata.update" && command.regenerateTitle === true) {
       yield* Ref.update(effects, (existing) => [
         ...existing,
@@ -14816,7 +14845,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
                 (cause) => new OrchestratorProjectionError({ threadId: deletion.threadId, cause }),
               ),
             );
-          return yield* mapDispatchError(command)(
+          const plan = yield* mapDispatchError(command)(
             planThreadDeletion({
               command: deletion,
               projection,
@@ -14827,6 +14856,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
               idAllocator,
             }),
           );
+          const cascade = yield* subagentCascadeEffects(deletion, deletion.type).pipe(
+            mapDispatchError(command),
+          );
+          return { ...plan, effects: [...plan.effects, ...cascade] };
         });
         // Checked under the thread lock: a message sent since the client looked keeps the thread.
         if (command.onlyIfUnused === true) {
@@ -15232,6 +15265,38 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       if (shell !== null) yield* refuseAboveDispatchModeLimit(command, threadId, shell);
     }
 
+    // Decided under the child's lock, so a newer independent choice on it always wins.
+    const cascade = yield* EffectOutbox.SubagentCascade;
+    if (
+      cascade !== undefined &&
+      (command.type === "thread.archive" ||
+        command.type === "thread.unarchive" ||
+        command.type === "thread.delete")
+    ) {
+      const { source, path } = cascade;
+      const read = mapDispatchError(command);
+      const thread = yield* projectionStore.getThread(command.threadId).pipe(read);
+      const parentThreadId = thread.lineage.parentThreadId;
+      const parent =
+        parentThreadId === null
+          ? null
+          : yield* projectionStore.getThread(parentThreadId).pipe(read);
+      // A child moved to another project since the walk found it is no longer this subtree's.
+      let applies = parent?.projectId === thread.projectId;
+      if (applies && source !== undefined && command.type !== "thread.delete") {
+        const latest = yield* commandReceipts.getLatestLifecycle(command.threadId).pipe(read);
+        applies =
+          thread.deletedAt === null &&
+          (command.type === "thread.archive"
+            ? thread.archivedAt === null && !supersedesCascade(latest, source, command.commandId)
+            : thread.archivedAt !== null &&
+              (yield* archivedByCascade(commandReceipts, latest, source.threadId, path).pipe(
+                read,
+              )));
+      }
+      if (!applies) return { sequence: source?.resultSequence ?? 0, storedEvents: [] };
+    }
+
     const plan = yield* dispatchOnce(command, sideChatIds).pipe(
       Effect.tap((planned) => refuseAboveRuntimeModeCeiling(command, planned.events)),
       Effect.filterOrFail(
@@ -16068,6 +16133,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
           .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause })));
       },
     ),
+    getSubagentChildThreads: Effect.fn("OrchestratorV2.getSubagentChildThreads")((threadId) =>
+      projectionStore
+        .getSubagentChildThreads(threadId)
+        .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))),
+    ),
     getThreadProjection: Effect.fn("OrchestratorV2.getThreadProjection")(function (threadId) {
       return projectionStore
         .getThreadProjection(threadId)

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

@@ -610,6 +610,10 @@ export interface ProjectionStoreV2Shape {
   readonly getThread: (
     threadId: ThreadId,
   ) => Effect.Effect<OrchestrationV2AppThread, ProjectionStoreV2Error>;
+  /** Direct subagents, archived and deleted included. */
+  readonly getSubagentChildThreads: (
+    threadId: ThreadId,
+  ) => Effect.Effect<ReadonlyArray<OrchestrationV2AppThread>, ProjectionStoreV2Error>;
   readonly getLimitRecoveryCandidates: (options: {
     readonly now: DateTime.Utc;
     readonly autoResume: boolean;
@@ -5575,6 +5579,20 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         ),
     );
 
+    // Subagents inherit the parent's immutable project. Use its index to bound the scan.
+    const getSubagentChildThreads: ProjectionStoreV2Shape["getSubagentChildThreads"] = (threadId) =>
+      sql<PayloadRow>`
+        SELECT child.payload_json FROM orchestration_v2_projection_threads AS child
+        JOIN orchestration_v2_projection_threads AS parent
+          ON parent.thread_id = ${threadId} AND child.project_id = parent.project_id
+        WHERE json_extract(child.payload_json, '$.lineage.parentThreadId') = ${threadId}
+          AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent'
+        ORDER BY child.created_at ASC, child.thread_id ASC
+      `.pipe(
+        Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })),
+        Effect.flatMap(decodeRows(decodeThreadPayload, threadId)),
+      );
+
     const requireThread = Effect.fn("ProjectionStore.requireThread")(function* (
       threadId: ThreadId,
     ) {
@@ -9081,6 +9099,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       getThreadLinkMetadata: readOnReader(getThreadLinkMetadata),
       getThreadWorktreePaths: readOnReader(getThreadWorktreePaths),
       getOpenRemoteDelegatedTasks: getOpenRemoteDelegatedTasks.pipe(NodeSqliteClient.readOnly),
+      getSubagentChildThreads: readOnReader(getSubagentChildThreads),
       getThreadProjection: readOnReader(getThreadProjection),
       getTurnStartContext: readOnReader(getTurnStartContext),
       getTurnStartHistory: readOnReader(getTurnStartHistory),

apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts

@@ -224,6 +224,7 @@ it.effect(
           getThreadShell: () => Effect.die("unused getThreadShell"),
           getThreadTitles: () => Effect.die("unused getThreadTitles"),
           getThreadLinkMetadata: () => Effect.die("unused getThreadLinkMetadata"),
+          getSubagentChildThreads: () => Effect.die("unused getSubagentChildThreads"),
           findGuestShareThreadId: () => Effect.die("unused findGuestShareThreadId"),
           getThread: () => Ref.get(projection).pipe(Effect.map((state) => state.thread)),
           getSettlementCandidates: () => Effect.die("unused getSettlementCandidates"),