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 bbb214542540d..345d7846ebd15 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/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({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c6d1ab4654110..b7c760ebb0b21 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; @@ -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) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3c1d353f061b4..c506a60387309 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, ) { @@ -9081,6 +9099,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..3e75e13fc3225 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"]; @@ -806,8 +815,34 @@ const make = Effect.gen(function* () { ensureCommandTranscripts(command).pipe( Effect.andThen(assertDispatchTargetEnabled(command)), Effect.andThen(orchestrator.dispatch(command)), + Effect.tap((result) => cascadeBeforeReply(command, result)), ); + // Clients read archive snapshots as soon as this returns; the outbox effect only backs it up. + // 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, + 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, @@ -1655,6 +1690,69 @@ 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; + readonly projectId?: 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 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; + 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 +1847,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..c3e7d82554b61 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,259 @@ 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 orchestrator = yield* Orchestrator.OrchestratorV2; + 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 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("")), + ); + 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, commit, 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 orchestrator = yield* Orchestrator.OrchestratorV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + // "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 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), stale, (c, t) => [...c, t]) : children(id), + ); + 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 }); + // Settled before the reply, without the effect worker, so a client refresh sees it. + yield* f.run(operation); + const field = operation === "delete" ? "deletedAt" : "archivedAt"; + assert.equal( + yield* f.state(field, ...all), + operation === "unarchive" ? "0000000" : "1110000", + ); + 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("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; + // 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.commit("archive"); + // Newer choices whose IDs only look like this cascade's still win. + yield* f.commit("unarchive", parent, "archive:parent:subagent:parent"); + yield* f.drain; + assert.isTrue(Option.isNone(yield* receipt("archive:parent:subagent:app"))); + yield* f.commit("archive", app); + yield* f.drain; + 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.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.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.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)), + ); + 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.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); + // 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.