From d6654e616f1dcd57ff5cdc8a6289e23fead50ae8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lu=C3=ADs=20Miguel?= Date: Sat, 14 Nov 2026 12:23:14 +0000 Subject: [PATCH 1/5] fix(server): cascade parent lifecycle to subagent descendants MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserve independent archives and later choices while walking existing subagent trees through one durable effect and ordinary lifecycle commands. Rebased replacement of closed, unmerged PR #20072 on main 13f89ae0a7. Preserves contributor and maintainer work from the original submission. Co-authored-by: Julius Marminge Co-authored-by: maria-rcks Co-authored-by: Claude Opus 5.5 > 🤖 Generated by GPT 6.1 Sol in Codex via T3 Code. --- .../orchestration-v2/CommandReceiptStore.ts | 69 ++++++ .../src/orchestration-v2/EffectOutbox.ts | 17 ++ .../src/orchestration-v2/EffectWorker.ts | 19 ++ .../src/orchestration-v2/Orchestrator.ts | 61 ++++- .../src/orchestration-v2/ProjectionStore.ts | 19 ++ .../ProviderTurnControlService.test.ts | 1 + .../ThreadManagementService.ts | 68 ++++++ .../src/orchestration-v2/runtimeLayer.test.ts | 216 +++++++++++++++++- .../OrchestrationCommandReceipts.ts | 27 +++ .../src/relay/AgentAwarenessRelay.test.ts | 1 + docs/user/thread-sidebar.md | 2 + 11 files changed, 497 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/CommandReceiptStore.ts b/apps/server/src/orchestration-v2/CommandReceiptStore.ts index f39de342b5526..7c2b12cdcfec9 100644 --- a/apps/server/src/orchestration-v2/CommandReceiptStore.ts +++ b/apps/server/src/orchestration-v2/CommandReceiptStore.ts @@ -31,9 +31,19 @@ export class CommandReceiptStoreReadError extends Schema.TaggedError()( + "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; readonly insertIfAbsent: ( receipt: AnyCommandReceiptV2, ) => Effect.Effect; @@ -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, +) { + 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; +}); diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index eb1209158cc74..e06caea31210f 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/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 } + | 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; diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 9dd5079805745..c1a7db26ed641 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -27,6 +27,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"; @@ -593,6 +594,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({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 2870e22ea5000..2311b12cdd01e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/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; + readonly getSubagentChildThreads: ( + ...args: Parameters + ) => Effect.Effect, OrchestratorV2Error>; readonly getCheckpointContext: ( threadId: ThreadId, ) => Effect.Effect; @@ -1773,6 +1781,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) { @@ -5164,6 +5186,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, @@ -14768,7 +14797,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, @@ -14779,6 +14808,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) { @@ -15184,6 +15217,25 @@ 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?.source !== undefined && + (command.type === "thread.archive" || command.type === "thread.unarchive") + ) { + const { source, path } = cascade; + const read = mapDispatchError(command); + const thread = yield* projectionStore.getThread(command.threadId).pipe(read); + const latest = yield* commandReceipts.getLatestLifecycle(command.threadId).pipe(read); + const 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, storedEvents: [] }; + } + const plan = yield* dispatchOnce(command, sideChatIds).pipe( Effect.tap((planned) => refuseAboveRuntimeModeCeiling(command, planned.events)), Effect.filterOrFail( @@ -16020,6 +16072,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) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 6d845450c79da..0675d20b21eaa 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -610,6 +610,10 @@ export interface ProjectionStoreV2Shape { readonly getThread: ( threadId: ThreadId, ) => Effect.Effect; + /** Direct subagents, archived and deleted included. */ + readonly getSubagentChildThreads: ( + threadId: ThreadId, + ) => Effect.Effect, ProjectionStoreV2Error>; readonly getLimitRecoveryCandidates: (options: { readonly now: DateTime.Utc; readonly autoResume: boolean; @@ -5575,6 +5579,20 @@ export const layer: Layer.Layer = ), ); + // Subagents inherit the parent's immutable project. Use its index to bound the scan. + const getSubagentChildThreads: ProjectionStoreV2Shape["getSubagentChildThreads"] = (threadId) => + sql` + 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, ) { @@ -9074,6 +9092,7 @@ export const layer: Layer.Layer = getThreadLinkMetadata: readOnReader(getThreadLinkMetadata), getThreadWorktreePaths: readOnReader(getThreadWorktreePaths), getOpenRemoteDelegatedTasks: getOpenRemoteDelegatedTasks.pipe(NodeSqliteClient.readOnly), + getSubagentChildThreads: readOnReader(getSubagentChildThreads), getThreadProjection: readOnReader(getThreadProjection), getTurnStartContext: readOnReader(getTurnStartContext), getTurnStartHistory: readOnReader(getTurnStartHistory), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index c3d1096cb3c93..37640e530363d 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/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"), diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 482674b772d43..0c1fede3b64aa 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -58,6 +58,7 @@ import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; +import * as EffectOutbox from "./EffectOutbox.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -514,6 +515,14 @@ export interface ThreadManagementServiceShape { void, Orchestrator.OrchestratorV2Error | ThreadManagementProviderInstanceDisabledError >; + readonly cascadeSubagentLifecycle: (input: { + readonly threadId: ThreadId; + readonly commandId: CommandId; + readonly operation: "thread.archive" | "thread.unarchive" | "thread.delete"; + }) => Effect.Effect< + void, + Orchestrator.OrchestratorV2Error | CommandReceiptStore.CommandReceiptStoreV2Error + >; readonly getThreadEventSequence: Orchestrator.OrchestratorV2["Service"]["getThreadEventSequence"]; readonly recoverDelegatedTask: Orchestrator.OrchestratorV2["Service"]["recoverDelegatedTask"]; readonly cancelDelegatedTask: Orchestrator.OrchestratorV2["Service"]["cancelDelegatedTask"]; @@ -1655,6 +1664,64 @@ const make = Effect.gen(function* () { if (failures[0] !== undefined) return yield* failures[0]; }); + const cascadeSubagentLifecycle: ThreadManagementServiceShape["cascadeSubagentLifecycle"] = + Effect.fn("ThreadManagementService.cascadeSubagentLifecycle")(function* (input) { + const receipt = Option.getOrUndefined(yield* receipts.getByCommandId(input.commandId)); + // Side chats deleted with their parent have no receipt of their own. + if (input.operation !== "thread.delete" && receipt?.status !== "accepted") return; + const source = input.operation === "thread.delete" ? undefined : receipt; + const visit: ( + thread: { readonly id: ThreadId; readonly deletedAt: unknown }, + commandId: CommandId, + path: ReadonlyArray, + ) => ReturnType = Effect.fn( + "ThreadManagementService.visitSubagentCascade", + )(function* (thread, commandId, path) { + if (path.length > 0 && thread.deletedAt === null) { + const dispatched = yield* orchestrator + .dispatch({ type: input.operation, commandId, threadId: thread.id }) + .pipe( + Effect.provideService(EffectOutbox.SubagentCascade, { source, path }), + Effect.as(true), + Effect.catchTags({ + OrchestratorThreadAboveModeLimitError: () => Effect.succeed(false), + }), + ); + if (!dispatched) return; + } + // Holds each thread's lock while visiting its subagents, so locks nest parent before child. + yield* threadDispatch.withLock( + thread.id, + Effect.gen(function* () { + // A deleted thread, a thread above the caller's limit or a newer choice keeps its subtree. + const limit = yield* DispatchModeLimit; + const current = (yield* orchestrator.getThreadRecords(thread.id, [])).thread; + if (limit && exceededDispatchModeLimit(limit, current)) return; + if (source && current.deletedAt !== null) return; + const latest = source ? yield* receipts.getLatestLifecycle(thread.id) : null; + if (source && CommandReceiptStore.supersedesCascade(latest, source, commandId)) return; + const children = yield* orchestrator.getSubagentChildThreads(thread.id); + const [, failures] = yield* Effect.partition(children, (child) => + Effect.gen(function* () { + const base = CommandId.make(`${commandId}:subagent:${child.id}`); + const childCommandId = yield* CommandReceiptStore.cascadeCommandId(receipts, base); + if (childCommandId !== undefined) + return yield* visit(child, childCommandId, [...path, child.id]); + return yield* new Orchestrator.OrchestratorDispatchError({ + commandId: base, + commandType: input.operation, + cause: "All 32 subagent lifecycle command IDs were rejected.", + }); + }), + ); + // Finish healthy siblings before reporting a failure to the durable worker. + if (failures[0] !== undefined) return yield* failures[0]; + }), + ); + }); + yield* visit({ id: input.threadId, deletedAt: null }, input.commandId, []); + }); + return ThreadManagementService.of({ ensureLegacyTranscript, searchThreadStream: (input) => @@ -1749,6 +1816,7 @@ const make = Effect.gen(function* () { settleThread, interruptThread, stopDelegatedTasks, + cascadeSubagentLifecycle, stopWorkflow, forceStopThread, awaitBackgroundTasks, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9e054e3a0bf89..472c8a67be0cb 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -6,7 +6,7 @@ import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderS import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { assert, it } from "@effect/vitest"; +import { assert, describe, it } from "@effect/vitest"; import { vi } from "vite-plus/test"; import { type ApplicationStoredEvent, @@ -60,6 +60,8 @@ import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ServerSettings from "../serverSettings.ts"; +import * as Receipts from "./CommandReceiptStore.ts"; +import { DispatchModeLimit } from "./DispatchModeLimit.ts"; import * as McpSessionRegistryTestkit from "../mcp/McpSessionRegistry.testkit.ts"; import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; @@ -346,6 +348,7 @@ const layerTest = Layer.mergeAll( ProjectStore.layer, ProjectionStore.layer, EffectOutbox.layer, + Receipts.layer, ThreadCommandExecutor.layer, ).pipe( Layer.provide(McpSessionRegistryTestkit.layer), @@ -11637,3 +11640,214 @@ it.layer(layerQueueResumeDisabledInstanceTest)( ); }, ); + +describe("subagent lifecycle cascade", () => { + const parent = ThreadId.make("parent"); + const app = ThreadId.make("app"); + const native = ThreadId.make("native"); + const grandchild = ThreadId.make("grandchild:retry:31"); + const fixture = Effect.gen(function* () { + const threads = yield* ThreadManagementService.ThreadManagementService; + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + yield* threads.dispatch({ + type: "thread.create", + commandId: CommandId.make("create:parent"), + threadId: parent, + projectId: ProjectId.make("project"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "agent", + creationSource: "mcp", + }); + const base = yield* store.getThread(parent); + const set = Effect.fn("test.seed")(function* (id: ThreadId, p: Partial) { + const thread = yield* store.getThread(id).pipe(Effect.orElseSucceed(() => base)); + const event = { id: EventId.make(`seed:${id}`), threadId: id, occurredAt: now }; + yield* store.apply({ ...event, type: "thread.created", payload: { ...thread, id, ...p } }); + }); + for (const [id, parentThreadId, relationshipToParent, projectId] of [ + [app, parent, "subagent", base.projectId], + [native, parent, "subagent", base.projectId], + [grandchild, app, "subagent", base.projectId], + [ThreadId.make("fork"), parent, "fork", base.projectId], + [ThreadId.make("fork-child"), ThreadId.make("fork"), "subagent", base.projectId], + [ThreadId.make("foreign"), parent, "subagent", ProjectId.make("foreign")], + ] as const) { + const lineage = { rootThreadId: parent, parentThreadId, relationshipToParent }; + yield* set(id, { lineage, projectId, creationSource: id === native ? "provider" : "mcp" }); + } + const run = (operation: "archive" | "unarchive" | "delete", threadId = parent, id?: string) => + threads.dispatch({ + type: `thread.${operation}`, + commandId: CommandId.make(id ?? `${operation}:${threadId}`), + threadId, + }); + const state = (field: "archivedAt" | "deletedAt", ...ids: ReadonlyArray) => + Effect.forEach(ids, (id) => store.getThread(ThreadId.make(id))).pipe( + Effect.map((all) => all.map((thread) => (thread[field] === null ? 0 : 1)).join("")), + ); + const archived = (...ids: ReadonlyArray) => state("archivedAt", ...ids); + const outbox = yield* EffectOutbox.EffectOutboxV2; + const cascade = Effect.map( + outbox.get("effect:archive:parent:subagent-threads.cascade"), + Option.getOrThrow, + ); + const drain = (yield* EffectWorker.OrchestrationEffectWorkerV2).drain(); + return { threads, now, set, run, state, archived, outbox, cascade, drain }; + }); + const receipt = (id: string) => + Effect.flatMap(Receipts.CommandReceiptStoreV2, (r) => r.getByCommandId(CommandId.make(id))); + it.effect("archives, restores and deletes subagents recursively, but not forks or projects", () => + Effect.gen(function* () { + const f = yield* fixture; + const all = [native, app, grandchild, "fork", "fork-child", "foreign"]; + for (const operation of ["archive", "unarchive", "delete"] as const) { + // A deleted intermediate must not hide surviving descendants from deletion. + if (operation === "delete") yield* f.set(app, { deletedAt: f.now }); + yield* f.run(operation); + yield* f.drain; + const field = operation === "delete" ? "deletedAt" : "archivedAt"; + assert.equal( + yield* f.state(field, ...all), + operation === "unarchive" ? "000000" : "111000", + ); + const effects = yield* f.outbox.listByCommandId( + CommandId.make(`${operation}:parent:subagent:native`), + ); + if (operation !== "unarchive") + assert.isTrue(effects.some((e) => e.request.type === "terminal.cleanup")); + } + }).pipe(Effect.provide(layerTest)), + ); + it.effect("restores only what a parent archive cascaded, through independent archives", () => + Effect.gen(function* () { + const f = yield* fixture; + // An earlier independent archive whose ID matches the one the parent cascade derives. + yield* f.run("archive", app, "archive:parent:subagent:app"); + yield* f.drain; + yield* f.run("archive"); + yield* f.drain; + yield* f.run("unarchive"); + yield* f.drain; + assert.equal(yield* f.archived(app, grandchild, native), "110"); + yield* f.run("unarchive", grandchild); + yield* f.run("archive", parent, "archive-owned"); + yield* f.drain; + yield* f.run("unarchive", native); + // An unused retry-shaped user ID is not a cascade retry. + yield* f.run("archive", native, "archive-owned:subagent:native:retry:31"); + yield* f.run("unarchive", parent, "unarchive-superseded"); + yield* f.run("archive", parent, "archive-superseding"); + yield* f.drain; + yield* TestClock.adjust("8 days"); + assert.isAbove(yield* f.outbox.pruneSettled, 0); + yield* f.run("unarchive", parent, "unarchive-latest"); + yield* f.drain; + assert.equal(yield* f.archived(app, native, grandchild), "110"); + }).pipe(Effect.provide(layerTest)), + ); + it.effect("skips a superseded parent action and keeps descendant choices made after it", () => + Effect.gen(function* () { + const f = yield* fixture; + yield* f.run("archive"); + // Newer choices whose IDs only look like this cascade's still win. + yield* f.run("unarchive", parent, "archive:parent:subagent:parent"); + yield* f.drain; + assert.isTrue(Option.isNone(yield* receipt("archive:parent:subagent:app"))); + yield* f.run("archive", app); + yield* f.drain; + yield* f.run("unarchive", grandchild); + yield* f.run("archive", parent, "archive-later"); + yield* f.run("unarchive", app, "archive-later:subagent:app:retry:31"); + yield* f.drain; + assert.equal(yield* f.archived(parent, native, app, grandchild), "1100"); + yield* f.run("unarchive", parent, "unarchive-later"); + yield* f.run("archive", app, "archive-app-later"); + yield* f.drain; + assert.equal(yield* f.archived(parent, native, app, grandchild), "0011"); + yield* f.set(app, { deletedAt: f.now }); + for (const operation of ["archive", "unarchive"] as const) { + yield* f.set(grandchild, { archivedAt: operation === "archive" ? null : f.now }); + yield* f.run(operation, parent, `${operation}-tombstone`).pipe(Effect.andThen(f.drain)); + assert.equal(yield* f.archived(grandchild), operation === "archive" ? "0" : "1"); + } + // A root tombstoned without its own receipt, like a side chat deleted with its parent. + yield* f.run("archive", parent, "archive-deleted"); + yield* f.set(parent, { deletedAt: f.now }).pipe(Effect.andThen(f.drain)); + assert.isTrue(Option.isNone(yield* receipt("archive-deleted:subagent:native"))); + }).pipe(Effect.provide(layerTest)), + ); + it.effect.each([31, 32])("bounds child retries at 32 IDs with %i rejected", (count) => + Effect.gen(function* () { + const f = yield* fixture; + const receipts = yield* Receipts.CommandReceiptStoreV2; + const created = Option.getOrThrow(yield* receipt("create:parent")); + for (let retry = 0; retry < count; retry++) { + const id = CommandId.make(`archive:parent:subagent:app${retry ? `:retry:${retry}` : ""}`); + yield* receipts.upsert({ ...created, commandId: id, threadId: app, status: "rejected" }); + } + yield* f.run("archive"); + // Lose only the outbox acknowledgement, after the child commands commit. + vi.spyOn(f.outbox, "succeed").mockImplementationOnce(() => Effect.succeed(true)); + yield* f.drain; + const accepted = yield* receipt("archive:parent:subagent:app:retry:31"); + if (count === 32) { + const { status, attemptCount, lastError } = yield* f.cascade; + assert.deepEqual([status, attemptCount], ["pending", 1]); + assert.include(lastError, "All 32 subagent lifecycle command IDs were rejected."); + } else { + assert.equal(Option.getOrThrow(accepted).status, "accepted"); + assert.equal((yield* f.cascade).status, "running"); + yield* f.run("unarchive", app); + yield* f.drain; + assert.equal((yield* f.outbox.reconcileAfterProcessLoss).requeued, 1); + yield* f.drain; + assert.equal((yield* f.cascade).status, "succeeded"); + assert.deepEqual(yield* receipt("archive:parent:subagent:app:retry:31"), accepted); + } + assert.equal(yield* f.archived(native, app, grandchild), "100"); + assert.isTrue(Option.isNone(yield* receipt("archive:parent:subagent:app:retry:32"))); + }).pipe(Effect.provide(layerTest)), + ); + it.effect.each([ + { type: "thread.runtime-mode.set", runtimeMode: "full-access", alreadyArchived: false }, + { type: "thread.interaction-mode.set", interactionMode: "default", alreadyArchived: true }, + ] as const)("keeps the caller's limit after $type", ({ alreadyArchived, ...raise }) => + Effect.gen(function* () { + const f = yield* fixture; + const commandId = CommandId.make("raise"); + const limit = { runtimeMode: "approval-required", interactionMode: "plan" } as const; + for (const id of [parent, app, native, grandchild]) yield* f.set(id, limit); + if (alreadyArchived) { + // The child's archive was accepted before the process was lost, so recovery replays it. + const receipts = yield* Receipts.CommandReceiptStoreV2; + const created = Option.getOrThrow(yield* receipt("create:parent")); + const id = CommandId.make("archive:parent:subagent:app"); + yield* receipts.upsert({ ...created, commandId: id, threadId: app }); + yield* f.set(app, { archivedAt: f.now }); + } + yield* f.run("archive").pipe(Effect.provideService(DispatchModeLimit, limit)); + const claimed = yield* f.outbox.claimNext({ workerId: "lost-caller" }); + assert.equal(Option.getOrThrow(claimed).request.type, "subagent-threads.cascade"); + assert.equal((yield* f.outbox.reconcileAfterProcessLoss).requeued, 1); + // The raise commits just before the walk takes app's lock: for dispatch, or after a replay. + const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const { withLock } = executor; + const raised = Effect.orDie(f.threads.dispatch({ ...raise, commandId, threadId: app })); + let requests = 0; + const lock: typeof withLock = (id, effect) => + id === app && ++requests === (alreadyArchived ? 2 : 1) + ? raised.pipe(Effect.andThen(withLock(id, effect))) + : withLock(id, effect); + using _lock = vi.spyOn(executor, "withLock").mockImplementation(lock); + yield* f.drain; + assert.equal(yield* f.archived(native, app, grandchild), alreadyArchived ? "110" : "100"); + assert.equal((yield* f.cascade).status, "succeeded"); + }).pipe(Effect.provide(layerTest)), + ); +}); diff --git a/apps/server/src/persistence/OrchestrationCommandReceipts.ts b/apps/server/src/persistence/OrchestrationCommandReceipts.ts index 4a39061a00cb0..0d20e04c49497 100644 --- a/apps/server/src/persistence/OrchestrationCommandReceipts.ts +++ b/apps/server/src/persistence/OrchestrationCommandReceipts.ts @@ -41,6 +41,13 @@ export class OrchestrationCommandReceiptRepository extends Context.Service< receipt: OrchestrationCommandReceipt, ) => Effect.Effect; + readonly getLatestLifecycle: (input: { + readonly threadId: ThreadId; + }) => Effect.Effect< + Option.Option, + OrchestrationCommandReceiptRepositoryError + >; + readonly getByCommandId: ( input: GetByCommandIdInput, ) => Effect.Effect< @@ -108,6 +115,20 @@ const makeOrchestrationCommandReceiptRepository = Effect.gen(function* () { `, }); + const findLatestLifecycle = SqlSchema.findOneOption({ + Request: Schema.Struct({ threadId: ThreadId }), + Result: OrchestrationCommandReceipt, + execute: ({ threadId }) => sql` + SELECT command_id AS "commandId", aggregate_kind AS "aggregateKind", + aggregate_id AS "aggregateId", command_type AS "commandType", + accepted_at AS "acceptedAt", result_sequence AS "resultSequence", status, error + FROM orchestration_command_receipts + WHERE aggregate_kind = 'thread' AND aggregate_id = ${threadId} + AND status = 'accepted' AND command_type IN ('thread.archive', 'thread.unarchive', 'thread.delete') + ORDER BY result_sequence DESC LIMIT 1 + `, + }); + const upsert: OrchestrationCommandReceiptRepository["Service"]["upsert"] = (receipt) => upsertReceiptRow(receipt).pipe( Effect.mapError(toPersistenceSqlError("OrchestrationCommandReceiptRepository.upsert:query")), @@ -157,6 +178,12 @@ const makeOrchestrationCommandReceiptRepository = Effect.gen(function* () { ); return { + getLatestLifecycle: (input) => + findLatestLifecycle(input).pipe( + Effect.mapError( + toPersistenceSqlError("OrchestrationCommandReceiptRepository.getLatestLifecycle:query"), + ), + ), insertIfAbsent, upsert, getByCommandId, diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 68e2289524c09..939dd3abf6666 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -260,6 +260,7 @@ const makeTestRelay = Effect.fnUntraced(function* ( forceStopThread: () => Effect.die("unused forceStopThread"), awaitBackgroundTasks: unused, stopDelegatedTasks: unused, + cascadeSubagentLifecycle: unused, getThreadEventSequence: unused, recoverDelegatedTask: unused, cancelDelegatedTask: unused, diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 4c02f1c79baee..de1681827497c 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -276,6 +276,8 @@ fill the screen; close the full-screen view to return to the parent and inspecto Stop on a thread also stops the subagents it delegated to. Before archiving the parent, wait for its delegated tasks to finish or cancel them. +Archiving, restoring or deleting a thread also applies to its subagent descendants, but not to forks. Restoring brings back only what the parent's archive archived, so independent archives and newer changes on a descendant stay as they are. + Subagent threads started by the agent can't take messages; message the parent thread instead. When such a subagent needs an approval or an answer, the parent thread asks for it. From eddab8c4248a61b9e1a438af115d6579800d8cc0 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 14:36:15 +0000 Subject: [PATCH 2/5] fix(server): finish subagent cascade before lifecycle commands reply The Archive page refreshes its snapshot as soon as archive, restore or delete returns, but the cascade ran later in the effect worker, so descendants showed their old state until a reload. Run the cascade before the reply and keep the outbox effect as the durable backstop. Co-Authored-By: Claude Opus 5.5 --- .../ThreadManagementService.ts | 20 +++++++++ .../src/orchestration-v2/runtimeLayer.test.ts | 42 ++++++++++++------- 2 files changed, 46 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 0c1fede3b64aa..b7bfb24c316fc 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -815,8 +815,28 @@ const make = Effect.gen(function* () { ensureCommandTranscripts(command).pipe( Effect.andThen(assertDispatchTargetEnabled(command)), Effect.andThen(orchestrator.dispatch(command)), + Effect.tap(() => cascadeBeforeReply(command)), ); + // Clients read archive snapshots as soon as this returns; the outbox effect only backs it up. + const cascadeBeforeReply = (command: OrchestrationV2ServerCommand) => + command.type === "thread.archive" || + command.type === "thread.unarchive" || + command.type === "thread.delete" + ? cascadeSubagentLifecycle({ + commandId: command.commandId, + threadId: command.threadId, + operation: command.type, + }).pipe( + Effect.catch((cause) => + Effect.logWarning("Subagent lifecycle cascade left to the effect worker", { + threadId: command.threadId, + cause, + }), + ), + ) + : Effect.void; + const getProjectThreadRecords: ThreadManagementServiceShape["getProjectThreadRecords"] = ( input, fields, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 472c8a67be0cb..77840672d3259 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -11648,6 +11648,7 @@ describe("subagent lifecycle cascade", () => { const grandchild = ThreadId.make("grandchild:retry:31"); const fixture = Effect.gen(function* () { const threads = yield* ThreadManagementService.ThreadManagementService; + const orchestrator = yield* Orchestrator.OrchestratorV2; const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; yield* threads.dispatch({ @@ -11681,12 +11682,21 @@ describe("subagent lifecycle cascade", () => { const lineage = { rootThreadId: parent, parentThreadId, relationshipToParent }; yield* set(id, { lineage, projectId, creationSource: id === native ? "provider" : "mcp" }); } - const run = (operation: "archive" | "unarchive" | "delete", threadId = parent, id?: string) => - threads.dispatch({ + const command = ( + operation: "archive" | "unarchive" | "delete", + threadId: ThreadId, + id?: string, + ) => + ({ type: `thread.${operation}`, commandId: CommandId.make(id ?? `${operation}:${threadId}`), threadId, - }); + }) as const; + const run = (operation: "archive" | "unarchive" | "delete", threadId = parent, id?: string) => + threads.dispatch(command(operation, threadId, id)); + // Committed by a process that stopped before its reply, leaving the cascade to the worker. + const commit = (operation: "archive" | "unarchive", threadId = parent, id?: string) => + orchestrator.dispatch(command(operation, threadId, id)); const state = (field: "archivedAt" | "deletedAt", ...ids: ReadonlyArray) => Effect.forEach(ids, (id) => store.getThread(ThreadId.make(id))).pipe( Effect.map((all) => all.map((thread) => (thread[field] === null ? 0 : 1)).join("")), @@ -11698,7 +11708,7 @@ describe("subagent lifecycle cascade", () => { Option.getOrThrow, ); const drain = (yield* EffectWorker.OrchestrationEffectWorkerV2).drain(); - return { threads, now, set, run, state, archived, outbox, cascade, drain }; + return { threads, now, set, run, commit, state, archived, outbox, cascade, drain }; }); const receipt = (id: string) => Effect.flatMap(Receipts.CommandReceiptStoreV2, (r) => r.getByCommandId(CommandId.make(id))); @@ -11709,8 +11719,8 @@ describe("subagent lifecycle cascade", () => { for (const operation of ["archive", "unarchive", "delete"] as const) { // A deleted intermediate must not hide surviving descendants from deletion. if (operation === "delete") yield* f.set(app, { deletedAt: f.now }); + // Settled before the reply, without the effect worker, so a client refresh sees it. yield* f.run(operation); - yield* f.drain; const field = operation === "delete" ? "deletedAt" : "archivedAt"; assert.equal( yield* f.state(field, ...all), @@ -11754,30 +11764,30 @@ describe("subagent lifecycle cascade", () => { it.effect("skips a superseded parent action and keeps descendant choices made after it", () => Effect.gen(function* () { const f = yield* fixture; - yield* f.run("archive"); + yield* f.commit("archive"); // Newer choices whose IDs only look like this cascade's still win. - yield* f.run("unarchive", parent, "archive:parent:subagent:parent"); + yield* f.commit("unarchive", parent, "archive:parent:subagent:parent"); yield* f.drain; assert.isTrue(Option.isNone(yield* receipt("archive:parent:subagent:app"))); - yield* f.run("archive", app); + yield* f.commit("archive", app); yield* f.drain; - yield* f.run("unarchive", grandchild); - yield* f.run("archive", parent, "archive-later"); - yield* f.run("unarchive", app, "archive-later:subagent:app:retry:31"); + yield* f.commit("unarchive", grandchild); + yield* f.commit("archive", parent, "archive-later"); + yield* f.commit("unarchive", app, "archive-later:subagent:app:retry:31"); yield* f.drain; assert.equal(yield* f.archived(parent, native, app, grandchild), "1100"); - yield* f.run("unarchive", parent, "unarchive-later"); - yield* f.run("archive", app, "archive-app-later"); + yield* f.commit("unarchive", parent, "unarchive-later"); + yield* f.commit("archive", app, "archive-app-later"); yield* f.drain; assert.equal(yield* f.archived(parent, native, app, grandchild), "0011"); yield* f.set(app, { deletedAt: f.now }); for (const operation of ["archive", "unarchive"] as const) { yield* f.set(grandchild, { archivedAt: operation === "archive" ? null : f.now }); - yield* f.run(operation, parent, `${operation}-tombstone`).pipe(Effect.andThen(f.drain)); + yield* f.commit(operation, parent, `${operation}-tombstone`).pipe(Effect.andThen(f.drain)); assert.equal(yield* f.archived(grandchild), operation === "archive" ? "0" : "1"); } // A root tombstoned without its own receipt, like a side chat deleted with its parent. - yield* f.run("archive", parent, "archive-deleted"); + yield* f.commit("archive", parent, "archive-deleted"); yield* f.set(parent, { deletedAt: f.now }).pipe(Effect.andThen(f.drain)); assert.isTrue(Option.isNone(yield* receipt("archive-deleted:subagent:native"))); }).pipe(Effect.provide(layerTest)), @@ -11831,7 +11841,7 @@ describe("subagent lifecycle cascade", () => { yield* receipts.upsert({ ...created, commandId: id, threadId: app }); yield* f.set(app, { archivedAt: f.now }); } - yield* f.run("archive").pipe(Effect.provideService(DispatchModeLimit, limit)); + yield* f.commit("archive").pipe(Effect.provideService(DispatchModeLimit, limit)); const claimed = yield* f.outbox.claimNext({ workerId: "lost-caller" }); assert.equal(Option.getOrThrow(claimed).request.type, "subagent-threads.cascade"); assert.equal((yield* f.outbox.reconcileAfterProcessLoss).requeued, 1); From 52da8cf3227c9efd6132d0a008f234f40f2ef384 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 14:45:33 +0000 Subject: [PATCH 3/5] fix(server): skip cascading to a subagent moved to another project The walk finds same-project children before it dispatches to them, so a child adopted into another project in between was still archived or deleted. Recheck the parent's project under the child's lock. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 33 +++++++++++++------ .../src/orchestration-v2/runtimeLayer.test.ts | 10 ++++++ 2 files changed, 33 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 4520491a1929b..b7c760ebb0b21 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -15268,20 +15268,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Decided under the child's lock, so a newer independent choice on it always wins. const cascade = yield* EffectOutbox.SubagentCascade; if ( - cascade?.source !== undefined && - (command.type === "thread.archive" || command.type === "thread.unarchive") + 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 latest = yield* commandReceipts.getLatestLifecycle(command.threadId).pipe(read); - const 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, storedEvents: [] }; + 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( diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 77840672d3259..b095a58e3ebcd 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -11715,6 +11715,16 @@ describe("subagent lifecycle cascade", () => { it.effect("archives, restores and deletes subagents recursively, but not forks or projects", () => Effect.gen(function* () { const f = yield* fixture; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + // "foreign" moved to another project after the walk found it. + const children = orchestrator.getSubagentChildThreads; + const foreign = store.getThread(ThreadId.make("foreign")).pipe(Effect.orDie); + using _stale = vi + .spyOn(orchestrator, "getSubagentChildThreads") + .mockImplementation((id) => + id === parent ? Effect.zipWith(children(id), foreign, (c, t) => [...c, t]) : children(id), + ); const all = [native, app, grandchild, "fork", "fork-child", "foreign"]; for (const operation of ["archive", "unarchive", "delete"] as const) { // A deleted intermediate must not hide surviving descendants from deletion. From 3dbe87c21d260f986199f3d05204d53497603905 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 14:51:44 +0000 Subject: [PATCH 4/5] fix(server): cascade before reply only for a committed lifecycle change A replayed command would walk the subtree under the retry's mode limit, and an unused-only delete that kept the parent would still delete its descendants. Leave both to the committed outbox cascade. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ThreadManagementService.ts | 16 +++++++++++----- .../src/orchestration-v2/runtimeLayer.test.ts | 12 ++++++++++++ 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index b7bfb24c316fc..cce138e13d0a8 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -815,14 +815,20 @@ const make = Effect.gen(function* () { ensureCommandTranscripts(command).pipe( Effect.andThen(assertDispatchTargetEnabled(command)), Effect.andThen(orchestrator.dispatch(command)), - Effect.tap(() => cascadeBeforeReply(command)), + Effect.tap((result) => cascadeBeforeReply(command, result)), ); // Clients read archive snapshots as soon as this returns; the outbox effect only backs it up. - const cascadeBeforeReply = (command: OrchestrationV2ServerCommand) => - command.type === "thread.archive" || - command.type === "thread.unarchive" || - command.type === "thread.delete" + // A replay or a no-op (an unused-only delete that kept the thread) leaves it to that effect. + const cascadeBeforeReply = ( + command: OrchestrationV2ServerCommand, + result: Orchestrator.OrchestratorV2DispatchResult, + ) => + (command.type === "thread.archive" || + command.type === "thread.unarchive" || + command.type === "thread.delete") && + result.replayed !== true && + result.storedEvents.length > 0 ? cascadeSubagentLifecycle({ commandId: command.commandId, threadId: command.threadId, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b095a58e3ebcd..9306ea781d52f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -11744,6 +11744,18 @@ describe("subagent lifecycle cascade", () => { } }).pipe(Effect.provide(layerTest)), ); + it.effect("leaves a replayed parent command's descendants to its committed cascade", () => + Effect.gen(function* () { + const f = yield* fixture; + const limit = { runtimeMode: "approval-required", interactionMode: "plan" } as const; + yield* f.set(parent, limit); + yield* f.run("archive").pipe(Effect.provideService(DispatchModeLimit, limit)); + assert.equal(yield* f.archived(parent, native, app), "100"); + // A retry without the caller's limit must not walk the subtree under its own. + yield* f.run("archive"); + assert.equal(yield* f.archived(native, app), "00"); + }).pipe(Effect.provide(layerTest)), + ); it.effect("restores only what a parent archive cascaded, through independent archives", () => Effect.gen(function* () { const f = yield* fixture; From d885a46d269b97f542e6105eca4ecd653a385a26 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 15:00:48 +0000 Subject: [PATCH 5/5] fix(server): stop the cascade at a subagent moved to another project A moved child was skipped, but the walk still visited its subagents, and one that moved with it passed the parent-project check. Stop the walk under the child's lock when its project changed since it was found. Co-Authored-By: Claude Opus 5.5 --- .../ThreadManagementService.ts | 9 ++++++-- .../src/orchestration-v2/runtimeLayer.test.ts | 23 +++++++++++++++---- 2 files changed, 25 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index cce138e13d0a8..3e75e13fc3225 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1697,7 +1697,11 @@ const make = Effect.gen(function* () { if (input.operation !== "thread.delete" && receipt?.status !== "accepted") return; const source = input.operation === "thread.delete" ? undefined : receipt; const visit: ( - thread: { readonly id: ThreadId; readonly deletedAt: unknown }, + thread: { + readonly id: ThreadId; + readonly deletedAt: unknown; + readonly projectId?: unknown; + }, commandId: CommandId, path: ReadonlyArray, ) => ReturnType = Effect.fn( @@ -1719,9 +1723,10 @@ const make = Effect.gen(function* () { yield* threadDispatch.withLock( thread.id, Effect.gen(function* () { - // A deleted thread, a thread above the caller's limit or a newer choice keeps its subtree. + // A deleted or moved thread, one above the caller's limit or a newer choice keeps its subtree. const limit = yield* DispatchModeLimit; const current = (yield* orchestrator.getThreadRecords(thread.id, [])).thread; + if (path.length > 0 && current.projectId !== thread.projectId) return; if (limit && exceededDispatchModeLimit(limit, current)) return; if (source && current.deletedAt !== null) return; const latest = source ? yield* receipts.getLatestLifecycle(thread.id) : null; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9306ea781d52f..c3e7d82554b61 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -11717,15 +11717,28 @@ describe("subagent lifecycle cascade", () => { const f = yield* fixture; const orchestrator = yield* Orchestrator.OrchestratorV2; const store = yield* ProjectionStore.ProjectionStoreV2; - // "foreign" moved to another project after the walk found it. + // "foreign" moved to another project, with its own subagent, after the walk found it. + const { projectId } = yield* store.getThread(parent); + const foreign = ThreadId.make("foreign"); + yield* f.set(ThreadId.make("foreign-child"), { + lineage: { + rootThreadId: parent, + parentThreadId: foreign, + relationshipToParent: "subagent", + }, + projectId: ProjectId.make("foreign"), + }); const children = orchestrator.getSubagentChildThreads; - const foreign = store.getThread(ThreadId.make("foreign")).pipe(Effect.orDie); + const stale = store.getThread(foreign).pipe( + Effect.orDie, + Effect.map((thread) => ({ ...thread, projectId })), + ); using _stale = vi .spyOn(orchestrator, "getSubagentChildThreads") .mockImplementation((id) => - id === parent ? Effect.zipWith(children(id), foreign, (c, t) => [...c, t]) : children(id), + id === parent ? Effect.zipWith(children(id), stale, (c, t) => [...c, t]) : children(id), ); - const all = [native, app, grandchild, "fork", "fork-child", "foreign"]; + const all = [native, app, grandchild, "fork", "fork-child", "foreign", "foreign-child"]; for (const operation of ["archive", "unarchive", "delete"] as const) { // A deleted intermediate must not hide surviving descendants from deletion. if (operation === "delete") yield* f.set(app, { deletedAt: f.now }); @@ -11734,7 +11747,7 @@ describe("subagent lifecycle cascade", () => { const field = operation === "delete" ? "deletedAt" : "archivedAt"; assert.equal( yield* f.state(field, ...all), - operation === "unarchive" ? "000000" : "111000", + operation === "unarchive" ? "0000000" : "1110000", ); const effects = yield* f.outbox.listByCommandId( CommandId.make(`${operation}:parent:subagent:native`),