From dd604f2b09297e8420e5392fdf5944ccd3427c83 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Mon, 19 Oct 2026 18:23:09 +0000 Subject: [PATCH 1/3] fix(server): provider subagent threads follow their parent into the archive Archiving or deleting a thread never archived the provider-native subagent threads (Claude Task/Agent children and the like) mirrored under it. Every client hides those threads, but they stayed in every active shell read: on bkt3 on 2026-10-19, 318 of 1,144 active threads were finished subagents of an archived parent. A fork-owned reactor (SubagentArchiveCascade.expbkt3.ts) now: - archives a parent's provider-native subagent children on thread.archived and thread.deleted, and unarchives them on thread.unarchived; - once per start, after activation, archives every provider-native subagent whose parent is archived, deleted or missing, and logs the count. delegate_task children, forks and children with an unfinished run are left alone. Candidates come from one index-backed SQL read; commands go out four at a time. server.ts gains one marked import and one marked layer line. Co-Authored-By: Claude Opus 5.5 --- .../SubagentArchiveCascade.expbkt3.test.ts | 357 ++++++++++++++++++ .../SubagentArchiveCascade.expbkt3.ts | 247 ++++++++++++ apps/server/src/server.ts | 2 + 3 files changed, 606 insertions(+) create mode 100644 apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts create mode 100644 apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts new file mode 100644 index 0000000000000..dd5dfc2fd7ee0 --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts @@ -0,0 +1,357 @@ +// T3-CUSTOM(expbkt3): provider-native subagents follow their parent into and out of the archive. +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + NodeId, + type OrchestrationV2AppThread, + type OrchestrationV2StoredEvent, + ProjectId, + ProviderInstanceId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { EventSinkV2 } from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import { ProviderAdapterRegistryV2 } from "./ProviderAdapterRegistry.ts"; +import * as SubagentArchiveCascade from "./SubagentArchiveCascade.expbkt3.ts"; +import { layerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; + +const layerTest = layerWithRegistry( + { name: "SubagentArchiveCascade.expbkt3.test.ts" }, + Layer.mock(ProviderAdapterRegistryV2)({ + get: () => Effect.die("Provider execution is outside this test."), + list: () => Effect.succeed([ProviderInstanceId.make("codex")]), + }), + { runEffectWorker: false, databaseLayer: SqlitePersistence.layerMemory }, +).pipe(Layer.provideMerge(SqlitePersistence.layerMemory)); + +const projectId = ProjectId.make("project:subagent-archive-cascade"); +const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }; + +const makeThread = ( + id: string, + now: DateTime.Utc, + overrides: Partial = {}, +): OrchestrationV2AppThread => { + const threadId = ThreadId.make(id); + return { + createdBy: "user", + creationSource: "web", + id: threadId, + projectId, + title: id, + providerInstanceId: modelSelection.instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + ...overrides, + }; +}; + +/** A child as each creator records it: a provider's native subagent, delegate_task, or a fork. */ +const makeChild = ( + kind: "provider-subagent" | "delegated" | "fork", + parentThreadId: ThreadId, + id: string, + now: DateTime.Utc, + overrides: Partial = {}, +) => + makeThread(id, now, { + createdBy: kind === "fork" ? "user" : "agent", + creationSource: + kind === "provider-subagent" ? "provider" : kind === "delegated" ? "mcp" : "web", + lineage: { + parentThreadId, + relationshipToParent: kind === "fork" ? "fork" : "subagent", + rootThreadId: parentThreadId, + }, + forkedFrom: kind === "fork" ? null : { type: "node", nodeId: NodeId.make(`node:${id}`) }, + ...overrides, + }); + +const seed = (threads: ReadonlyArray) => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + yield* eventSink.write({ + events: threads.map((thread) => ({ + id: EventId.make(`event:${thread.id}:created`), + type: "thread.created" as const, + threadId: thread.id, + occurredAt: thread.createdAt, + payload: thread, + })), + }); + }); + +const seedRunningRun = (threadId: ThreadId, now: DateTime.Utc) => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const runId = RunId.make(`run:${threadId}:1`); + yield* eventSink.write({ + events: [ + { + id: EventId.make(`event:${runId}:created`), + type: "run.created", + threadId, + runId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: 1, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make(`message:${runId}`), + rootNodeId: null, + activeAttemptId: null, + status: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }, + ], + }); + }); + +/** The next stored event of this type on the thread, even one that lands before the read subscribes. */ +const nextEvent = (threadId: ThreadId, eventType: "thread.archived" | "thread.unarchived") => + Effect.gen(function* () { + const eventSink = yield* EventSinkV2; + const afterSequence = yield* eventSink.latestSequence(); + return yield* eventSink + .stream({ threadId, afterSequence, eventType }) + .pipe(Stream.runHead, Effect.forkChild); + }); + +const commandIdOf = (stored: Option.Option) => + String(Option.getOrNull(Option.map(stored, (event) => event.commandId))); + +const isArchived = (threadId: ThreadId) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const shell = yield* orchestrator.getThreadShell(threadId); + return shell !== null && shell.archivedAt !== null; + }); + +it.effect( + "archiving a parent archives its provider subagents, not its delegated tasks, forks or other threads", + () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:parent", now); + const other = makeThread("thread:other", now); + const subagent = makeChild("provider-subagent", parent.id, "thread:parent:subagent", now); + const delegated = makeChild("delegated", parent.id, "thread:parent:delegated", now); + const fork = makeChild("fork", parent.id, "thread:parent:fork", now); + const otherSubagent = makeChild( + "provider-subagent", + other.id, + "thread:other:subagent", + now, + ); + yield* seed([parent, other, subagent, delegated, fork, otherSubagent]); + + const cascade = yield* SubagentArchiveCascade.make; + yield* cascade.start(); + // Every parent is active, so the startup pass archives nothing. + yield* cascade.drain; + assert.isFalse(yield* isArchived(subagent.id)); + + const subagentArchived = yield* nextEvent(subagent.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-parent"), + threadId: parent.id, + }); + const stored = yield* Fiber.join(subagentArchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-archive-cascade:")); + assert.isTrue(yield* isArchived(parent.id)); + assert.isTrue(yield* isArchived(subagent.id)); + assert.isFalse(yield* isArchived(delegated.id)); + assert.isFalse(yield* isArchived(fork.id)); + assert.isFalse(yield* isArchived(other.id)); + assert.isFalse(yield* isArchived(otherSubagent.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("deleting a parent archives its provider subagents", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:deleted-parent", now); + const subagent = makeChild( + "provider-subagent", + parent.id, + "thread:deleted-parent:subagent", + now, + ); + yield* seed([parent, subagent]); + + const cascade = yield* SubagentArchiveCascade.make; + yield* cascade.start(); + yield* cascade.drain; + + const subagentArchived = yield* nextEvent(subagent.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("command:delete-parent"), + threadId: parent.id, + }); + const stored = yield* Fiber.join(subagentArchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-archive-cascade:")); + assert.isNull(yield* orchestrator.getThreadShell(parent.id)); + assert.isTrue(yield* isArchived(subagent.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("unarchiving the parent unarchives its provider subagents and nothing else", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:round-trip", now); + const subagent = makeChild("provider-subagent", parent.id, "thread:round-trip:subagent", now); + // Archived by the user before the parent was, so the parent's unarchive must not restore it. + const delegated = makeChild("delegated", parent.id, "thread:round-trip:delegated", now, { + archivedAt: now, + }); + yield* seed([parent, subagent, delegated]); + + const cascade = yield* SubagentArchiveCascade.make; + yield* cascade.start(); + yield* cascade.drain; + + const subagentArchived = yield* nextEvent(subagent.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-round-trip"), + threadId: parent.id, + }); + yield* Fiber.join(subagentArchived); + yield* cascade.drain; + assert.isTrue(yield* isArchived(subagent.id)); + + const subagentUnarchived = yield* nextEvent(subagent.id, "thread.unarchived"); + yield* orchestrator.dispatch({ + type: "thread.unarchive", + commandId: CommandId.make("command:unarchive-round-trip"), + threadId: parent.id, + }); + const stored = yield* Fiber.join(subagentUnarchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-unarchive-cascade:")); + assert.isFalse(yield* isArchived(parent.id)); + assert.isFalse(yield* isArchived(subagent.id)); + assert.isTrue(yield* isArchived(delegated.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect( + "the startup pass archives subagents whose parent is archived, deleted or gone, and only those", + () => + Effect.scoped( + Effect.gen(function* () { + const now = yield* DateTime.now; + const archivedParent = makeThread("thread:archived-parent", now, { archivedAt: now }); + const deletedParent = makeThread("thread:gone-parent", now, { deletedAt: now }); + const activeParent = makeThread("thread:active-parent", now); + const ofArchived = makeChild( + "provider-subagent", + archivedParent.id, + "thread:archived-parent:subagent", + now, + ); + const ofDeleted = makeChild( + "provider-subagent", + deletedParent.id, + "thread:gone-parent:subagent", + now, + ); + const ofMissing = makeChild( + "provider-subagent", + ThreadId.make("thread:never-recorded"), + "thread:never-recorded:subagent", + now, + ); + const stillRunning = makeChild( + "provider-subagent", + archivedParent.id, + "thread:archived-parent:running-subagent", + now, + ); + const delegated = makeChild( + "delegated", + archivedParent.id, + "thread:archived-parent:delegated", + now, + ); + const ofActive = makeChild( + "provider-subagent", + activeParent.id, + "thread:active-parent:subagent", + now, + ); + yield* seed([ + archivedParent, + deletedParent, + activeParent, + ofArchived, + ofDeleted, + ofMissing, + stillRunning, + delegated, + ofActive, + ]); + yield* seedRunningRun(stillRunning.id, now); + + const cascade = yield* SubagentArchiveCascade.make; + yield* cascade.start(); + yield* cascade.drain; + + assert.isTrue(yield* isArchived(ofArchived.id)); + assert.isTrue(yield* isArchived(ofDeleted.id)); + assert.isTrue(yield* isArchived(ofMissing.id)); + assert.isFalse(yield* isArchived(stillRunning.id)); + assert.isFalse(yield* isArchived(delegated.id)); + assert.isFalse(yield* isArchived(ofActive.id)); + assert.isFalse(yield* isArchived(activeParent.id)); + }), + ).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts new file mode 100644 index 0000000000000..0ba6fa5287d04 --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts @@ -0,0 +1,247 @@ +/** + * T3-CUSTOM(expbkt3): a provider-native subagent thread follows its parent into + * and out of the archive. + * + * Claude's Task/Agent tool children, and other providers' native subagents, are + * mirrored as threads (`isProviderNativeSubagentThread`) that every client + * hides. Archiving or deleting the parent never touched them, so a finished + * subagent stayed in every active shell read for good: on bkt3 on 2026-10-19, + * 318 of the 1,144 active threads were subagents of an archived parent. + * + * - `thread.archived` or `thread.deleted` of a parent archives its + * provider-native subagent children, and `thread.unarchived` unarchives them. + * - Once per start, after activation, every provider-native subagent whose + * parent is archived, deleted or gone is archived. That pass is the cleanup of + * the orphans that predate this rule, and heals a cascade a restart cut short. + * + * delegate_task children (`creationSource: "mcp"`) and forks keep their own + * lifecycle, and a child with an unfinished run is left alone. Both directions + * read where the parent is now, so a later event that reverses an earlier one + * wins. + */ +import { + CommandId, + isProviderNativeSubagentThread, + type OrchestrationV2DomainEvent, + type OrchestrationV2ServerCommand, + type ProjectId, + ThreadId, +} from "@t3tools/contracts"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; +import * as Cause from "effect/Cause"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as Stream from "effect/Stream"; + +import { forkParked, ServerActivation } from "../serverActivation.ts"; +import { EventSinkV2 } from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import { randomUuidV4 } from "./RandomUuid.ts"; + +/** Commands in flight at once, so a startup pass of hundreds does not crowd out live work. */ +const DISPATCH_CONCURRENCY = 4; + +type CascadeDirection = "archive" | "unarchive"; + +type CascadeRequest = + | { readonly type: "startup" } + | { + readonly type: "follow"; + readonly direction: CascadeDirection; + readonly parentThreadId: ThreadId; + readonly projectId: ProjectId; + }; + +interface ChildRow { + readonly thread_id: string; +} + +/** The cascade a domain event starts, or null when its children have nothing to follow. */ +function cascadeRequestFor(event: OrchestrationV2DomainEvent): CascadeRequest | null { + if ( + event.type !== "thread.archived" && + event.type !== "thread.unarchived" && + event.type !== "thread.deleted" + ) { + return null; + } + // A provider-native subagent has no provider-native children, so its own + // archive (most are sent from here) needs no lookup. + if (isProviderNativeSubagentThread(event.payload)) return null; + return { + type: "follow", + direction: event.type === "thread.unarchived" ? "unarchive" : "archive", + parentThreadId: event.threadId, + projectId: event.payload.projectId, + }; +} + +export const make = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSinkV2; + const sql = yield* SqlClient.SqlClient; + const activation = (yield* ServerActivation) ?? Effect.void; + + /** + * A parent's provider-native subagents that are not yet where the parent is. + * A child shares its parent's project, so the project index bounds the scan. + */ + const selectChildren = (request: Extract) => + sql` + SELECT child.thread_id AS thread_id + FROM orchestration_v2_projection_threads AS child + WHERE child.project_id = ${request.projectId} + AND child.deleted_at IS NULL + AND ${request.direction === "archive" ? sql`child.archived_at IS NULL` : sql`child.archived_at IS NOT NULL`} + AND CASE WHEN json_valid(child.payload_json) THEN + json_extract(child.payload_json, '$.lineage.parentThreadId') = ${request.parentThreadId} + AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' + AND json_extract(child.payload_json, '$.creationSource') = 'provider' + ELSE 0 END + AND ${request.direction === "archive" ? sql`NOT EXISTS` : sql`EXISTS`} ( + SELECT 1 + FROM orchestration_v2_projection_threads AS parent + WHERE parent.thread_id = ${request.parentThreadId} + AND parent.archived_at IS NULL + AND parent.deleted_at IS NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_runs AS run + WHERE run.thread_id = child.thread_id + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + ORDER BY child.created_at ASC, child.thread_id ASC + `; + + /** Unarchived provider-native subagents whose parent is archived, deleted or gone. */ + const selectOrphans = () => + sql` + SELECT child.thread_id AS thread_id + FROM orchestration_v2_projection_threads AS child + LEFT JOIN orchestration_v2_projection_threads AS parent + ON parent.thread_id = CASE WHEN json_valid(child.payload_json) + THEN json_extract(child.payload_json, '$.lineage.parentThreadId') END + WHERE child.deleted_at IS NULL + AND child.archived_at IS NULL + AND CASE WHEN json_valid(child.payload_json) THEN + json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL + AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' + AND json_extract(child.payload_json, '$.creationSource') = 'provider' + ELSE 0 END + AND ( + parent.thread_id IS NULL + OR parent.archived_at IS NOT NULL + OR parent.deleted_at IS NOT NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_runs AS run + WHERE run.thread_id = child.thread_id + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + ORDER BY child.created_at ASC, child.thread_id ASC + `; + + /** Dispatches one archive or unarchive per child; returns how many were accepted. */ + const moveChildren = ( + direction: CascadeDirection, + rows: ReadonlyArray, + source: "cascade" | "startup", + ) => + Effect.forEach( + rows, + (row) => + Effect.gen(function* () { + const threadId = ThreadId.make(row.thread_id); + const uuid = yield* randomUuidV4; + const commandId = CommandId.make( + `server:subagent-${direction}-${source}:${threadId}:${uuid}`, + ); + const command: OrchestrationV2ServerCommand = + direction === "archive" + ? { type: "thread.archive", commandId, threadId } + : { type: "thread.unarchive", commandId, threadId }; + yield* orchestrator.dispatch(command); + return true; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: a child did not follow its parent", { + threadId: row.thread_id, + direction, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), + ), + { concurrency: DISPATCH_CONCURRENCY }, + ).pipe(Effect.map((accepted) => accepted.filter(Boolean).length)); + + const processRequest = Effect.fn("SubagentArchiveCascade.process")(function* ( + request: CascadeRequest, + ) { + if (request.type === "startup") { + const orphans = yield* selectOrphans(); + const archived = yield* moveChildren("archive", orphans, "startup"); + yield* Effect.logInfo( + "subagent archive cascade: archived the subagents of archived or deleted parents", + { candidates: orphans.length, archived }, + ); + return; + } + const children = yield* selectChildren(request); + if (children.length === 0) return; + const moved = yield* moveChildren(request.direction, children, "cascade"); + yield* Effect.logInfo("subagent archive cascade: subagents followed their parent", { + parentThreadId: request.parentThreadId, + direction: request.direction, + candidates: children.length, + moved, + }); + }); + + // Items wait for activation, so the startup pass queued in `start` runs + // after the server is up and `drain` still covers it. + const worker = yield* makeDrainableWorker((request: CascadeRequest) => + activation.pipe( + Effect.andThen(processRequest(request)), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: request failed", { + request: request.type === "startup" ? "startup" : request.parentThreadId, + cause: Cause.pretty(cause), + }), + ), + ), + ); + + const start = Effect.fn("SubagentArchiveCascade.start")(function* () { + // Read from the current head: an event that lands before the parked + // stream subscribes is replayed rather than lost. + const afterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); + yield* worker.enqueue({ type: "startup" }); + yield* forkParked( + eventSink.stream({ afterSequence }).pipe( + Stream.runForEach((stored) => { + const request = cascadeRequestFor(stored.event); + return request === null ? Effect.void : worker.enqueue(request); + }), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: event stream failed", { + cause: Cause.pretty(cause), + }), + ), + ), + ); + }); + + return { start, drain: worker.drain }; +}); + +/** Starts the cascade with the server; one marked line in `server.ts` provides it. */ +export const layer = Layer.effectDiscard(Effect.flatMap(make, (cascade) => cascade.start())); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index b1383df1e47c0..b959420cdbc10 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -43,6 +43,7 @@ import * as SessionArchiveSweeper from "./sessionArchive/SessionArchiveSweeper.t import * as ThreadWorkspaceGroups from "./persistence/ThreadWorkspaceGroups.ts"; import { ArchiveExportReactorLive } from "./orchestration-v2/Layers/ArchiveExportReactor.ts"; import { ArchiveExportReactor } from "./orchestration-v2/Services/ArchiveExportReactor.ts"; +import * as SubagentArchiveCascade from "./orchestration-v2/SubagentArchiveCascade.expbkt3.ts"; import { ProjectionSnapshotQueryLive } from "./orchestration-v2/Layers/ProjectionSnapshotQuery.ts"; import { OrchestrationEngineLive } from "./orchestration-v2/Layers/OrchestrationEngine.ts"; import * as TurnStartBootstrap from "./orchestration-v2/turnStartBootstrap.expbkt3.ts"; @@ -658,6 +659,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( layerProviderInstallationRefresh, ReplayMarkers.layer, SessionActivityReactor.layer, // T3-CUSTOM(expbkt3): session time-in-state metrics (TEC-1631). + SubagentArchiveCascade.layer, // T3-CUSTOM(expbkt3): provider subagents follow their parent's archive. ).pipe( // Core Services Layer.provideMerge(layerOrchestrationApplication), From 17cec6bf0669d660fd8c12e953af65f487b1cc5a Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Mon, 19 Oct 2026 18:31:46 +0000 Subject: [PATCH 2/3] fix(server): pace the startup pass that archives orphaned subagents Every archive also starts a history export that runs git in the worktree, so archiving bkt3's ~318 orphaned subagents at once would burst ~1,300-1,600 git processes right after a deploy. The startup pass now queues one orphan per pause (default 2 seconds, set with `make({ startupPause })`), so the archives and their exports spread over ~11 minutes. A separate fiber does the pacing and feeds the same worker that runs live cascades. Live cascades therefore run between orphans and are not held for the whole pass, and nothing races them on the same child. Each orphan is re-checked when its turn comes and skipped if its parent came back. The pass logs progress once per 50 orphans, then one summary line. Tests pass a zero pause, and a new test advances TestClock to show one archive per pause. Co-Authored-By: Claude Opus 5.5 --- .../SubagentArchiveCascade.expbkt3.test.ts | 58 ++- .../SubagentArchiveCascade.expbkt3.ts | 380 ++++++++++-------- 2 files changed, 275 insertions(+), 163 deletions(-) diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts index dd5dfc2fd7ee0..ac4d4e61a2fd2 100644 --- a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts @@ -13,11 +13,13 @@ import { ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; import { EventSinkV2 } from "./EventSink.ts"; @@ -179,7 +181,7 @@ it.effect( ); yield* seed([parent, other, subagent, delegated, fork, otherSubagent]); - const cascade = yield* SubagentArchiveCascade.make; + const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); yield* cascade.start(); // Every parent is active, so the startup pass archives nothing. yield* cascade.drain; @@ -219,7 +221,7 @@ it.effect("deleting a parent archives its provider subagents", () => ); yield* seed([parent, subagent]); - const cascade = yield* SubagentArchiveCascade.make; + const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); yield* cascade.start(); yield* cascade.drain; @@ -252,7 +254,7 @@ it.effect("unarchiving the parent unarchives its provider subagents and nothing }); yield* seed([parent, subagent, delegated]); - const cascade = yield* SubagentArchiveCascade.make; + const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); yield* cascade.start(); yield* cascade.drain; @@ -341,7 +343,7 @@ it.effect( ]); yield* seedRunningRun(stillRunning.id, now); - const cascade = yield* SubagentArchiveCascade.make; + const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); yield* cascade.start(); yield* cascade.drain; @@ -355,3 +357,51 @@ it.effect( }), ).pipe(Effect.provide(layerTest)), ); + +it.effect("the startup pass archives one orphan per pause", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:paced-parent", now, { archivedAt: now }); + const first = makeChild("provider-subagent", parent.id, "thread:paced-parent:1", now); + const second = makeChild("provider-subagent", parent.id, "thread:paced-parent:2", now); + const third = makeChild("provider-subagent", parent.id, "thread:paced-parent:3", now); + yield* seed([parent, first, second, third]); + const firstArchived = yield* nextEvent(first.id, "thread.archived"); + const secondArchived = yield* nextEvent(second.id, "thread.archived"); + const thirdArchived = yield* nextEvent(third.id, "thread.archived"); + + const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.seconds(2) }); + yield* cascade.start(); + + // The first orphan goes at once; each later one waits for the clock to move a pause. + yield* Fiber.join(firstArchived); + assert.isFalse(yield* isArchived(second.id)); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(secondArchived); + assert.isFalse(yield* isArchived(third.id)); + yield* TestClock.adjust("2 seconds"); + yield* Fiber.join(thirdArchived); + yield* cascade.drain; + + const archivedAtMillis = (threadId: ThreadId) => + orchestrator + .getThreadShell(threadId) + .pipe( + Effect.map((shell) => + shell?.archivedAt == null ? null : DateTime.toEpochMillis(shell.archivedAt), + ), + ); + const start = DateTime.toEpochMillis(now); + assert.deepEqual( + [ + yield* archivedAtMillis(first.id), + yield* archivedAtMillis(second.id), + yield* archivedAtMillis(third.id), + ], + [start, start + 2_000, start + 4_000], + ); + }), + ).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts index 0ba6fa5287d04..c8e449a2d3c64 100644 --- a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts @@ -13,11 +13,12 @@ * - Once per start, after activation, every provider-native subagent whose * parent is archived, deleted or gone is archived. That pass is the cleanup of * the orphans that predate this rule, and heals a cascade a restart cut short. + * It archives one orphan per `startupPause`, because every archive also starts + * a history export (`ArchiveExportReactor`) that runs git in the worktree. * * delegate_task children (`creationSource: "mcp"`) and forks keep their own - * lifecycle, and a child with an unfinished run is left alone. Both directions - * read where the parent is now, so a later event that reverses an earlier one - * wins. + * lifecycle, and a child with an unfinished run is left alone. Every step reads + * where the parent is now, so a later event that reverses an earlier one wins. */ import { CommandId, @@ -29,6 +30,8 @@ import { } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; @@ -39,19 +42,26 @@ import { EventSinkV2 } from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { randomUuidV4 } from "./RandomUuid.ts"; -/** Commands in flight at once, so a startup pass of hundreds does not crowd out live work. */ +/** Commands in flight at once while one parent's children follow it. */ const DISPATCH_CONCURRENCY = 4; +/** ~318 startup archives on bkt3 spread over ~11 minutes, with their history exports. */ +const DEFAULT_STARTUP_PAUSE = Duration.seconds(2); + +/** The startup pass logs its progress once per this many orphans. */ +const STARTUP_PROGRESS_EVERY = 50; + type CascadeDirection = "archive" | "unarchive"; type CascadeRequest = - | { readonly type: "startup" } | { readonly type: "follow"; readonly direction: CascadeDirection; readonly parentThreadId: ThreadId; readonly projectId: ProjectId; - }; + } + | { readonly type: "startup-orphan"; readonly threadId: ThreadId } + | { readonly type: "startup-summary" }; interface ChildRow { readonly thread_id: string; @@ -77,171 +87,223 @@ function cascadeRequestFor(event: OrchestrationV2DomainEvent): CascadeRequest | }; } -export const make = Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSinkV2; - const sql = yield* SqlClient.SqlClient; - const activation = (yield* ServerActivation) ?? Effect.void; - - /** - * A parent's provider-native subagents that are not yet where the parent is. - * A child shares its parent's project, so the project index bounds the scan. - */ - const selectChildren = (request: Extract) => - sql` - SELECT child.thread_id AS thread_id - FROM orchestration_v2_projection_threads AS child - WHERE child.project_id = ${request.projectId} - AND child.deleted_at IS NULL - AND ${request.direction === "archive" ? sql`child.archived_at IS NULL` : sql`child.archived_at IS NOT NULL`} - AND CASE WHEN json_valid(child.payload_json) THEN - json_extract(child.payload_json, '$.lineage.parentThreadId') = ${request.parentThreadId} - AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' - AND json_extract(child.payload_json, '$.creationSource') = 'provider' - ELSE 0 END - AND ${request.direction === "archive" ? sql`NOT EXISTS` : sql`EXISTS`} ( - SELECT 1 - FROM orchestration_v2_projection_threads AS parent - WHERE parent.thread_id = ${request.parentThreadId} - AND parent.archived_at IS NULL - AND parent.deleted_at IS NULL - ) - AND NOT EXISTS ( - SELECT 1 - FROM orchestration_v2_projection_runs AS run - WHERE run.thread_id = child.thread_id - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') - ) - ORDER BY child.created_at ASC, child.thread_id ASC - `; - - /** Unarchived provider-native subagents whose parent is archived, deleted or gone. */ - const selectOrphans = () => - sql` - SELECT child.thread_id AS thread_id - FROM orchestration_v2_projection_threads AS child - LEFT JOIN orchestration_v2_projection_threads AS parent - ON parent.thread_id = CASE WHEN json_valid(child.payload_json) - THEN json_extract(child.payload_json, '$.lineage.parentThreadId') END - WHERE child.deleted_at IS NULL - AND child.archived_at IS NULL - AND CASE WHEN json_valid(child.payload_json) THEN - json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL - AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' - AND json_extract(child.payload_json, '$.creationSource') = 'provider' - ELSE 0 END - AND ( - parent.thread_id IS NULL - OR parent.archived_at IS NOT NULL - OR parent.deleted_at IS NOT NULL - ) - AND NOT EXISTS ( - SELECT 1 - FROM orchestration_v2_projection_runs AS run - WHERE run.thread_id = child.thread_id - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') - ) - ORDER BY child.created_at ASC, child.thread_id ASC - `; - - /** Dispatches one archive or unarchive per child; returns how many were accepted. */ - const moveChildren = ( - direction: CascadeDirection, - rows: ReadonlyArray, - source: "cascade" | "startup", - ) => - Effect.forEach( - rows, - (row) => - Effect.gen(function* () { - const threadId = ThreadId.make(row.thread_id); - const uuid = yield* randomUuidV4; - const commandId = CommandId.make( - `server:subagent-${direction}-${source}:${threadId}:${uuid}`, - ); - const command: OrchestrationV2ServerCommand = - direction === "archive" - ? { type: "thread.archive", commandId, threadId } - : { type: "thread.unarchive", commandId, threadId }; - yield* orchestrator.dispatch(command); - return true; - }).pipe( - Effect.catchCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.failCause(cause) - : Effect.logWarning("subagent archive cascade: a child did not follow its parent", { - threadId: row.thread_id, - direction, - cause: Cause.pretty(cause), - }).pipe(Effect.as(false)), - ), - ), - { concurrency: DISPATCH_CONCURRENCY }, - ).pipe(Effect.map((accepted) => accepted.filter(Boolean).length)); +export const make = (options: { readonly startupPause?: Duration.Input } = {}) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSinkV2; + const sql = yield* SqlClient.SqlClient; + const activation = (yield* ServerActivation) ?? Effect.void; + const startupPause = Duration.fromInputUnsafe(options.startupPause ?? DEFAULT_STARTUP_PAUSE); + // Completed once the startup pass has queued its last orphan. + const startupQueued = yield* Deferred.make(); + const startupPass = { candidates: 0, processed: 0, archived: 0 }; - const processRequest = Effect.fn("SubagentArchiveCascade.process")(function* ( - request: CascadeRequest, - ) { - if (request.type === "startup") { - const orphans = yield* selectOrphans(); - const archived = yield* moveChildren("archive", orphans, "startup"); - yield* Effect.logInfo( - "subagent archive cascade: archived the subagents of archived or deleted parents", - { candidates: orphans.length, archived }, + /** + * A parent's provider-native subagents that are not yet where the parent is. + * A child shares its parent's project, so the project index bounds the scan. + */ + const selectChildren = (request: Extract) => + sql` + SELECT child.thread_id AS thread_id + FROM orchestration_v2_projection_threads AS child + WHERE child.project_id = ${request.projectId} + AND child.deleted_at IS NULL + AND ${request.direction === "archive" ? sql`child.archived_at IS NULL` : sql`child.archived_at IS NOT NULL`} + AND CASE WHEN json_valid(child.payload_json) THEN + json_extract(child.payload_json, '$.lineage.parentThreadId') = ${request.parentThreadId} + AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' + AND json_extract(child.payload_json, '$.creationSource') = 'provider' + ELSE 0 END + AND ${request.direction === "archive" ? sql`NOT EXISTS` : sql`EXISTS`} ( + SELECT 1 + FROM orchestration_v2_projection_threads AS parent + WHERE parent.thread_id = ${request.parentThreadId} + AND parent.archived_at IS NULL + AND parent.deleted_at IS NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_runs AS run + WHERE run.thread_id = child.thread_id + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + ORDER BY child.created_at ASC, child.thread_id ASC + `; + + /** + * Unarchived provider-native subagents whose parent is archived, deleted or + * gone: all of them, or only `threadId` when it is still one. + */ + const selectOrphans = (threadId?: ThreadId) => + sql` + SELECT child.thread_id AS thread_id + FROM orchestration_v2_projection_threads AS child + LEFT JOIN orchestration_v2_projection_threads AS parent + ON parent.thread_id = CASE WHEN json_valid(child.payload_json) + THEN json_extract(child.payload_json, '$.lineage.parentThreadId') END + WHERE child.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND child.thread_id = ${threadId}`} + AND child.archived_at IS NULL + AND CASE WHEN json_valid(child.payload_json) THEN + json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL + AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' + AND json_extract(child.payload_json, '$.creationSource') = 'provider' + ELSE 0 END + AND ( + parent.thread_id IS NULL + OR parent.archived_at IS NOT NULL + OR parent.deleted_at IS NOT NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_runs AS run + WHERE run.thread_id = child.thread_id + AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') + ) + ORDER BY child.created_at ASC, child.thread_id ASC + `; + + /** Archives or unarchives one child; true when the orchestrator accepted it. */ + const moveChild = ( + direction: CascadeDirection, + threadId: ThreadId, + source: "cascade" | "startup", + ) => + Effect.gen(function* () { + const uuid = yield* randomUuidV4; + const commandId = CommandId.make( + `server:subagent-${direction}-${source}:${threadId}:${uuid}`, + ); + const command: OrchestrationV2ServerCommand = + direction === "archive" + ? { type: "thread.archive", commandId, threadId } + : { type: "thread.unarchive", commandId, threadId }; + yield* orchestrator.dispatch(command); + return true; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: a child did not follow its parent", { + threadId, + direction, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), ); - return; - } - const children = yield* selectChildren(request); - if (children.length === 0) return; - const moved = yield* moveChildren(request.direction, children, "cascade"); - yield* Effect.logInfo("subagent archive cascade: subagents followed their parent", { - parentThreadId: request.parentThreadId, - direction: request.direction, - candidates: children.length, - moved, + + const processRequest = Effect.fn("SubagentArchiveCascade.process")(function* ( + request: CascadeRequest, + ) { + switch (request.type) { + case "follow": { + const children = yield* selectChildren(request); + if (children.length === 0) return; + const accepted = yield* Effect.forEach( + children, + (child) => moveChild(request.direction, ThreadId.make(child.thread_id), "cascade"), + { concurrency: DISPATCH_CONCURRENCY }, + ); + yield* Effect.logInfo("subagent archive cascade: subagents followed their parent", { + parentThreadId: request.parentThreadId, + direction: request.direction, + candidates: children.length, + moved: accepted.filter(Boolean).length, + }); + return; + } + case "startup-orphan": { + // Queued up to minutes ago: skip it if its parent came back or it moved since. + const stillOrphaned = (yield* selectOrphans(request.threadId)).length > 0; + if (stillOrphaned && (yield* moveChild("archive", request.threadId, "startup"))) { + startupPass.archived += 1; + } + startupPass.processed += 1; + if ( + startupPass.processed % STARTUP_PROGRESS_EVERY === 0 && + startupPass.processed < startupPass.candidates + ) { + yield* Effect.logInfo("subagent archive cascade: startup pass progress", { + ...startupPass, + }); + } + return; + } + case "startup-summary": + yield* Effect.logInfo( + "subagent archive cascade: archived the subagents of archived or deleted parents", + { ...startupPass }, + ); + return; + } }); - }); - // Items wait for activation, so the startup pass queued in `start` runs - // after the server is up and `drain` still covers it. - const worker = yield* makeDrainableWorker((request: CascadeRequest) => - activation.pipe( - Effect.andThen(processRequest(request)), - Effect.catchCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.failCause(cause) - : Effect.logWarning("subagent archive cascade: request failed", { - request: request.type === "startup" ? "startup" : request.parentThreadId, - cause: Cause.pretty(cause), - }), - ), - ), - ); - - const start = Effect.fn("SubagentArchiveCascade.start")(function* () { - // Read from the current head: an event that lands before the parked - // stream subscribes is replayed rather than lost. - const afterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); - yield* worker.enqueue({ type: "startup" }); - yield* forkParked( - eventSink.stream({ afterSequence }).pipe( - Stream.runForEach((stored) => { - const request = cascadeRequestFor(stored.event); - return request === null ? Effect.void : worker.enqueue(request); - }), + // Items wait for activation. Live cascades and startup orphans share this + // one worker, so they never race each other on the same child. + const worker = yield* makeDrainableWorker((request: CascadeRequest) => + activation.pipe( + Effect.andThen(processRequest(request)), Effect.catchCause((cause) => Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) - : Effect.logWarning("subagent archive cascade: event stream failed", { + : Effect.logWarning("subagent archive cascade: request failed", { + request, cause: Cause.pretty(cause), }), ), ), ); - }); - return { start, drain: worker.drain }; -}); + /** Queues the orphans one per pause, so live cascades run between them. */ + const queueStartupPass = Effect.gen(function* () { + const orphans = yield* selectOrphans(); + startupPass.candidates = orphans.length; + for (const [index, orphan] of orphans.entries()) { + if (index > 0 && !Duration.isZero(startupPause)) yield* Effect.sleep(startupPause); + yield* worker.enqueue({ + type: "startup-orphan", + threadId: ThreadId.make(orphan.thread_id), + }); + } + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: startup pass failed", { + cause: Cause.pretty(cause), + }), + ), + Effect.andThen(worker.enqueue({ type: "startup-summary" })), + Effect.ensuring(Deferred.succeed(startupQueued, undefined)), + ); + + const start = Effect.fn("SubagentArchiveCascade.start")(function* () { + // Read from the current head: an event that lands before the parked + // stream subscribes is replayed rather than lost. + const afterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); + yield* forkParked( + eventSink.stream({ afterSequence }).pipe( + Stream.runForEach((stored) => { + const request = cascadeRequestFor(stored.event); + return request === null ? Effect.void : worker.enqueue(request); + }), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: event stream failed", { + cause: Cause.pretty(cause), + }), + ), + ), + ); + yield* forkParked(queueStartupPass); + }); + + return { + start, + /** After `start`: waits for the whole startup pass, then for every queued cascade. */ + drain: Deferred.await(startupQueued).pipe(Effect.andThen(worker.drain)), + }; + }); /** Starts the cascade with the server; one marked line in `server.ts` provides it. */ -export const layer = Layer.effectDiscard(Effect.flatMap(make, (cascade) => cascade.start())); +export const layer = Layer.effectDiscard(Effect.flatMap(make(), (cascade) => cascade.start())); From 63fd610cea95a11da391490a3d8f9f58ae35d71b Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Mon, 19 Oct 2026 19:00:42 +0000 Subject: [PATCH 3/3] fix(server): guard subagent archives on the subagent record and pace live cascades The "unfinished run" guard could never fire: a provider-native subagent thread has no run (0 of 777 on bkt3). Its liveness lives in its parent's subagent record (orchestration_v2_projection_subagents, one per child thread). An archive now skips a child whose record is pending, running or waiting, using the child_thread_id index. When a provider-native record stops being live, its thread is re-checked and follows an archived parent, so archiving a parent mid-turn no longer leaves its subagents behind until the next restart. Unarchive no longer applies the guard. Live cascades are now paced too: a live archive or unarchive waits until the previous live move is `cascadePause` old (default 500 ms, an option to `make`). A bulk clean-out of parents with 40-60 subagents each therefore no longer bursts history exports. Startup orphans still come one per `startupPause` from the pacing fiber, and they interleave with live work in the shared worker. Each child is re-read just before it moves. Tests seed a live subagent record instead of a run. New tests cover a subagent that finishes after its parent was archived, and a cascade paced on the TestClock. Co-Authored-By: Claude Opus 5.5 --- .../SubagentArchiveCascade.expbkt3.test.ts | 192 ++++++++++++++--- .../SubagentArchiveCascade.expbkt3.ts | 200 +++++++++++------- 2 files changed, 282 insertions(+), 110 deletions(-) diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts index ac4d4e61a2fd2..d44943873ab82 100644 --- a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts @@ -3,13 +3,12 @@ import { assert, it } from "@effect/vitest"; import { CommandId, EventId, - MessageId, NodeId, type OrchestrationV2AppThread, type OrchestrationV2StoredEvent, ProjectId, + ProviderDriverKind, ProviderInstanceId, - RunId, ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -39,6 +38,7 @@ const layerTest = layerWithRegistry( const projectId = ProjectId.make("project:subagent-archive-cascade"); const modelSelection = { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }; +const driver = ProviderDriverKind.make("codex"); const makeThread = ( id: string, @@ -107,34 +107,49 @@ const seed = (threads: ReadonlyArray) => }); }); -const seedRunningRun = (threadId: ThreadId, now: DateTime.Utc) => +/** + * The parent's record of a provider-native subagent, as the adapters write it: + * a native subagent's thread has no run, so this record says whether it is live. + */ +const seedSubagentRecord = ( + child: OrchestrationV2AppThread, + status: "running" | "completed", + now: DateTime.Utc, +) => Effect.gen(function* () { const eventSink = yield* EventSinkV2; - const runId = RunId.make(`run:${threadId}:1`); + const parentThreadId = child.lineage.parentThreadId!; + const subagentId = NodeId.make(`subagent:${child.id}`); yield* eventSink.write({ events: [ { - id: EventId.make(`event:${runId}:created`), - type: "run.created", - threadId, - runId, + id: EventId.make(`event:${subagentId}:${status}`), + type: "subagent.updated", + threadId: parentThreadId, + nodeId: subagentId, + driver, + providerInstanceId: modelSelection.instanceId, occurredAt: now, payload: { - id: runId, - threadId, - ordinal: 1, + id: subagentId, + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make(`node:${parentThreadId}:root`), + origin: "provider_native", + createdBy: "agent", + driver, providerInstanceId: modelSelection.instanceId, - modelSelection, providerThreadId: null, - userMessageId: MessageId.make(`message:${runId}`), - rootNodeId: null, - activeAttemptId: null, - status: "running", - requestedAt: now, + childThreadId: child.id, + nativeTaskRef: null, + prompt: "Explore the repository", + title: null, + model: null, + status, + result: status === "completed" ? "Done." : null, startedAt: now, - completedAt: null, - checkpointId: null, - contextHandoffId: null, + completedAt: status === "completed" ? now : null, + updatedAt: now, }, }, ], @@ -161,6 +176,15 @@ const isArchived = (threadId: ThreadId) => return shell !== null && shell.archivedAt !== null; }); +/** When the thread was archived, on the test clock. */ +const archivedAtMillis = (threadId: ThreadId) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const shell = yield* orchestrator.getThreadShell(threadId); + if (shell?.archivedAt == null) return yield* Effect.die(`${threadId} is not archived.`); + return DateTime.toEpochMillis(shell.archivedAt); + }); + it.effect( "archiving a parent archives its provider subagents, not its delegated tasks, forks or other threads", () => @@ -181,7 +205,10 @@ it.effect( ); yield* seed([parent, other, subagent, delegated, fork, otherSubagent]); - const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); yield* cascade.start(); // Every parent is active, so the startup pass archives nothing. yield* cascade.drain; @@ -221,7 +248,10 @@ it.effect("deleting a parent archives its provider subagents", () => ); yield* seed([parent, subagent]); - const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); yield* cascade.start(); yield* cascade.drain; @@ -254,7 +284,10 @@ it.effect("unarchiving the parent unarchives its provider subagents and nothing }); yield* seed([parent, subagent, delegated]); - const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); yield* cascade.start(); yield* cascade.drain; @@ -341,9 +374,12 @@ it.effect( delegated, ofActive, ]); - yield* seedRunningRun(stillRunning.id, now); + yield* seedSubagentRecord(stillRunning, "running", now); - const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.zero }); + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); yield* cascade.start(); yield* cascade.drain; @@ -361,7 +397,6 @@ it.effect( it.effect("the startup pass archives one orphan per pause", () => Effect.scoped( Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; const now = yield* DateTime.now; const parent = makeThread("thread:paced-parent", now, { archivedAt: now }); const first = makeChild("provider-subagent", parent.id, "thread:paced-parent:1", now); @@ -372,7 +407,10 @@ it.effect("the startup pass archives one orphan per pause", () => const secondArchived = yield* nextEvent(second.id, "thread.archived"); const thirdArchived = yield* nextEvent(third.id, "thread.archived"); - const cascade = yield* SubagentArchiveCascade.make({ startupPause: Duration.seconds(2) }); + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.seconds(2), + cascadePause: Duration.zero, + }); yield* cascade.start(); // The first orphan goes at once; each later one waits for the clock to move a pause. @@ -385,14 +423,6 @@ it.effect("the startup pass archives one orphan per pause", () => yield* Fiber.join(thirdArchived); yield* cascade.drain; - const archivedAtMillis = (threadId: ThreadId) => - orchestrator - .getThreadShell(threadId) - .pipe( - Effect.map((shell) => - shell?.archivedAt == null ? null : DateTime.toEpochMillis(shell.archivedAt), - ), - ); const start = DateTime.toEpochMillis(now); assert.deepEqual( [ @@ -405,3 +435,95 @@ it.effect("the startup pass archives one orphan per pause", () => }), ).pipe(Effect.provide(layerTest)), ); + +it.effect("a subagent still running when its parent is archived follows once it finishes", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:busy-parent", now); + // Read before its sibling, so the sibling's archive shows the cascade has passed it. + const running = makeChild( + "provider-subagent", + parent.id, + "thread:busy-parent:a-running", + now, + ); + const done = makeChild("provider-subagent", parent.id, "thread:busy-parent:b-done", now); + yield* seed([parent, running, done]); + yield* seedSubagentRecord(running, "running", now); + yield* seedSubagentRecord(done, "completed", now); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + yield* cascade.start(); + yield* cascade.drain; + + const doneArchived = yield* nextEvent(done.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-busy-parent"), + threadId: parent.id, + }); + yield* Fiber.join(doneArchived); + yield* cascade.drain; + assert.isFalse(yield* isArchived(running.id)); + + const runningArchived = yield* nextEvent(running.id, "thread.archived"); + yield* seedSubagentRecord(running, "completed", now); + const stored = yield* Fiber.join(runningArchived); + yield* cascade.drain; + + assert.isTrue(commandIdOf(stored).startsWith("server:subagent-archive-cascade:")); + assert.isTrue(yield* isArchived(running.id)); + }), + ).pipe(Effect.provide(layerTest)), +); + +it.effect("a parent's subagents follow it one per pause", () => + Effect.scoped( + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const parent = makeThread("thread:paced-cascade", now); + const first = makeChild("provider-subagent", parent.id, "thread:paced-cascade:1", now); + const second = makeChild("provider-subagent", parent.id, "thread:paced-cascade:2", now); + const third = makeChild("provider-subagent", parent.id, "thread:paced-cascade:3", now); + yield* seed([parent, first, second, third]); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.millis(500), + }); + yield* cascade.start(); + yield* cascade.drain; + + const firstArchived = yield* nextEvent(first.id, "thread.archived"); + const thirdArchived = yield* nextEvent(third.id, "thread.archived"); + yield* orchestrator.dispatch({ + type: "thread.archive", + commandId: CommandId.make("command:archive-paced-cascade"), + threadId: parent.id, + }); + yield* Fiber.join(firstArchived); + // The clock has not moved, so the next child still waits for its pause. + assert.isFalse(yield* isArchived(second.id)); + + // Move the clock in small steps until the last child lands; the bound turns a + // stuck cascade into a failure instead of a hang. + for (let step = 0; step < 1_000 && thirdArchived.pollUnsafe() === undefined; step += 1) { + yield* TestClock.adjust("50 millis"); + } + assert.isDefined(thirdArchived.pollUnsafe()); + yield* cascade.drain; + + const firstAt = yield* archivedAtMillis(first.id); + const secondAt = yield* archivedAtMillis(second.id); + const thirdAt = yield* archivedAtMillis(third.id); + assert.isAtLeast(secondAt - firstAt, 500); + assert.isAtLeast(thirdAt - secondAt, 500); + }), + ).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts index c8e449a2d3c64..a1af2fb4be24d 100644 --- a/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts @@ -10,15 +10,18 @@ * * - `thread.archived` or `thread.deleted` of a parent archives its * provider-native subagent children, and `thread.unarchived` unarchives them. + * - A child whose subagent record is still pending, running or waiting is not + * archived. It follows its parent when that record stops being live. * - Once per start, after activation, every provider-native subagent whose * parent is archived, deleted or gone is archived. That pass is the cleanup of * the orphans that predate this rule, and heals a cascade a restart cut short. - * It archives one orphan per `startupPause`, because every archive also starts - * a history export (`ArchiveExportReactor`) that runs git in the worktree. * - * delegate_task children (`creationSource: "mcp"`) and forks keep their own - * lifecycle, and a child with an unfinished run is left alone. Every step reads - * where the parent is now, so a later event that reverses an earlier one wins. + * Every archive also starts a history export (`ArchiveExportReactor`) that runs + * git in the worktree, so nothing here moves children in a burst: live moves are + * at least `cascadePause` apart, and the startup pass queues one orphan per + * `startupPause` between them. delegate_task children (`creationSource: "mcp"`) + * and forks keep their own lifecycle. Each child is re-read just before it + * moves, so a later event that reverses an earlier one wins. */ import { CommandId, @@ -30,6 +33,7 @@ import { } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -42,33 +46,52 @@ import { EventSinkV2 } from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; import { randomUuidV4 } from "./RandomUuid.ts"; -/** Commands in flight at once while one parent's children follow it. */ -const DISPATCH_CONCURRENCY = 4; - /** ~318 startup archives on bkt3 spread over ~11 minutes, with their history exports. */ const DEFAULT_STARTUP_PAUSE = Duration.seconds(2); +/** A parent with 63 subagents (bkt3's largest) archives them over ~30 seconds. */ +const DEFAULT_CASCADE_PAUSE = Duration.millis(500); + /** The startup pass logs its progress once per this many orphans. */ const STARTUP_PROGRESS_EVERY = 50; type CascadeDirection = "archive" | "unarchive"; type CascadeRequest = + /** A parent moved: its children follow. */ | { readonly type: "follow"; readonly direction: CascadeDirection; readonly parentThreadId: ThreadId; readonly projectId: ProjectId; } + /** A subagent record stopped being live: its thread follows an archived parent. */ + | { readonly type: "finished"; readonly threadId: ThreadId } | { readonly type: "startup-orphan"; readonly threadId: ThreadId } | { readonly type: "startup-summary" }; +/** Which children a read covers: every one, one parent's, or a single child. */ +type ChildScope = + | { readonly kind: "all" } + | { readonly kind: "parent"; readonly parentThreadId: ThreadId; readonly projectId: ProjectId } + | { readonly kind: "child"; readonly threadId: ThreadId }; + interface ChildRow { readonly thread_id: string; } -/** The cascade a domain event starts, or null when its children have nothing to follow. */ +/** The work a domain event starts, or null when no child has anything to follow. */ function cascadeRequestFor(event: OrchestrationV2DomainEvent): CascadeRequest | null { + if (event.type === "subagent.updated") { + const subagent = event.payload; + const live = + subagent.status === "pending" || + subagent.status === "running" || + subagent.status === "waiting"; + return subagent.origin === "provider_native" && subagent.childThreadId !== null && !live + ? { type: "finished", threadId: subagent.childThreadId } + : null; + } if ( event.type !== "thread.archived" && event.type !== "thread.unarchived" && @@ -87,88 +110,103 @@ function cascadeRequestFor(event: OrchestrationV2DomainEvent): CascadeRequest | }; } -export const make = (options: { readonly startupPause?: Duration.Input } = {}) => +export const make = ( + options: { + readonly startupPause?: Duration.Input; + readonly cascadePause?: Duration.Input; + } = {}, +) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const eventSink = yield* EventSinkV2; const sql = yield* SqlClient.SqlClient; const activation = (yield* ServerActivation) ?? Effect.void; const startupPause = Duration.fromInputUnsafe(options.startupPause ?? DEFAULT_STARTUP_PAUSE); + const cascadePauseMillis = Duration.toMillis(options.cascadePause ?? DEFAULT_CASCADE_PAUSE); // Completed once the startup pass has queued its last orphan. const startupQueued = yield* Deferred.make(); const startupPass = { candidates: 0, processed: 0, archived: 0 }; + let lastLiveMoveAt: number | null = null; /** - * A parent's provider-native subagents that are not yet where the parent is. - * A child shares its parent's project, so the project index bounds the scan. - */ - const selectChildren = (request: Extract) => - sql` - SELECT child.thread_id AS thread_id - FROM orchestration_v2_projection_threads AS child - WHERE child.project_id = ${request.projectId} - AND child.deleted_at IS NULL - AND ${request.direction === "archive" ? sql`child.archived_at IS NULL` : sql`child.archived_at IS NOT NULL`} - AND CASE WHEN json_valid(child.payload_json) THEN - json_extract(child.payload_json, '$.lineage.parentThreadId') = ${request.parentThreadId} - AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' - AND json_extract(child.payload_json, '$.creationSource') = 'provider' - ELSE 0 END - AND ${request.direction === "archive" ? sql`NOT EXISTS` : sql`EXISTS`} ( - SELECT 1 - FROM orchestration_v2_projection_threads AS parent - WHERE parent.thread_id = ${request.parentThreadId} - AND parent.archived_at IS NULL - AND parent.deleted_at IS NULL - ) - AND NOT EXISTS ( - SELECT 1 - FROM orchestration_v2_projection_runs AS run - WHERE run.thread_id = child.thread_id - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') - ) - ORDER BY child.created_at ASC, child.thread_id ASC - `; - - /** - * Unarchived provider-native subagents whose parent is archived, deleted or - * gone: all of them, or only `threadId` when it is still one. + * Provider-native subagents in `scope` that are not where their parent is. + * To archive: the child is unarchived, its parent is archived, deleted or + * gone, and its subagent record is not live. To unarchive: the child is + * archived and its parent is active. A child shares its parent's project, so + * a parent's read stays on the project index; a child's is a key lookup. */ - const selectOrphans = (threadId?: ThreadId) => + const selectChildren = (direction: CascadeDirection, scope: ChildScope) => sql` SELECT child.thread_id AS thread_id FROM orchestration_v2_projection_threads AS child LEFT JOIN orchestration_v2_projection_threads AS parent ON parent.thread_id = CASE WHEN json_valid(child.payload_json) THEN json_extract(child.payload_json, '$.lineage.parentThreadId') END - WHERE child.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND child.thread_id = ${threadId}`} - AND child.archived_at IS NULL + WHERE child.deleted_at IS NULL${ + scope.kind === "parent" + ? sql` AND child.project_id = ${scope.projectId}` + : scope.kind === "child" + ? sql` AND child.thread_id = ${scope.threadId}` + : sql`` + } + AND ${direction === "archive" ? sql`child.archived_at IS NULL` : sql`child.archived_at IS NOT NULL`} AND CASE WHEN json_valid(child.payload_json) THEN - json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL + json_extract(child.payload_json, '$.lineage.parentThreadId') ${ + scope.kind === "parent" ? sql`= ${scope.parentThreadId}` : sql`IS NOT NULL` + } AND json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent' AND json_extract(child.payload_json, '$.creationSource') = 'provider' ELSE 0 END - AND ( - parent.thread_id IS NULL - OR parent.archived_at IS NOT NULL - OR parent.deleted_at IS NOT NULL - ) - AND NOT EXISTS ( - SELECT 1 - FROM orchestration_v2_projection_runs AS run - WHERE run.thread_id = child.thread_id - AND run.status IN ('preparing', 'queued', 'starting', 'running', 'waiting') - ) + AND ${ + direction === "archive" + ? sql`( + parent.thread_id IS NULL + OR parent.archived_at IS NOT NULL + OR parent.deleted_at IS NOT NULL + ) + AND NOT EXISTS ( + SELECT 1 + FROM orchestration_v2_projection_subagents AS subagent + WHERE subagent.child_thread_id = child.thread_id + AND subagent.status IN ('pending', 'running', 'waiting') + )` + : sql`( + parent.thread_id IS NOT NULL + AND parent.archived_at IS NULL + AND parent.deleted_at IS NULL + )` + } ORDER BY child.created_at ASC, child.thread_id ASC `; - /** Archives or unarchives one child; true when the orchestrator accepted it. */ + /** Waits until the last live move is `cascadePause` old. */ + const paceLiveMove = Effect.gen(function* () { + if (lastLiveMoveAt === null) return; + const wait = lastLiveMoveAt + cascadePauseMillis - (yield* Clock.currentTimeMillis); + if (wait > 0) yield* Effect.sleep(Duration.millis(wait)); + }); + + const isDue = (direction: CascadeDirection, threadId: ThreadId) => + selectChildren(direction, { kind: "child", threadId }).pipe( + Effect.map((rows) => rows.length > 0), + ); + + /** + * Re-reads one child and moves it when it still needs to; true when the + * orchestrator accepted the command. A live move waits for its pace, then + * reads the child again, so a check with nothing to do never waits. + */ const moveChild = ( direction: CascadeDirection, threadId: ThreadId, source: "cascade" | "startup", ) => Effect.gen(function* () { + if (!(yield* isDue(direction, threadId))) return false; + if (source === "cascade" && lastLiveMoveAt !== null && cascadePauseMillis > 0) { + yield* paceLiveMove; + if (!(yield* isDue(direction, threadId))) return false; + } const uuid = yield* randomUuidV4; const commandId = CommandId.make( `server:subagent-${direction}-${source}:${threadId}:${uuid}`, @@ -178,6 +216,7 @@ export const make = (options: { readonly startupPause?: Duration.Input } = {}) = ? { type: "thread.archive", commandId, threadId } : { type: "thread.unarchive", commandId, threadId }; yield* orchestrator.dispatch(command); + if (source === "cascade") lastLiveMoveAt = yield* Clock.currentTimeMillis; return true; }).pipe( Effect.catchCause((cause) => @@ -196,25 +235,36 @@ export const make = (options: { readonly startupPause?: Duration.Input } = {}) = ) { switch (request.type) { case "follow": { - const children = yield* selectChildren(request); + const children = yield* selectChildren(request.direction, { + kind: "parent", + parentThreadId: request.parentThreadId, + projectId: request.projectId, + }); if (children.length === 0) return; - const accepted = yield* Effect.forEach( - children, - (child) => moveChild(request.direction, ThreadId.make(child.thread_id), "cascade"), - { concurrency: DISPATCH_CONCURRENCY }, - ); + let moved = 0; + for (const child of children) { + if (yield* moveChild(request.direction, ThreadId.make(child.thread_id), "cascade")) { + moved += 1; + } + } yield* Effect.logInfo("subagent archive cascade: subagents followed their parent", { parentThreadId: request.parentThreadId, direction: request.direction, candidates: children.length, - moved: accepted.filter(Boolean).length, + moved, }); return; } + case "finished": + if (yield* moveChild("archive", request.threadId, "cascade")) { + yield* Effect.logInfo( + "subagent archive cascade: a finished subagent followed its archived parent", + { threadId: request.threadId }, + ); + } + return; case "startup-orphan": { - // Queued up to minutes ago: skip it if its parent came back or it moved since. - const stillOrphaned = (yield* selectOrphans(request.threadId)).length > 0; - if (stillOrphaned && (yield* moveChild("archive", request.threadId, "startup"))) { + if (yield* moveChild("archive", request.threadId, "startup")) { startupPass.archived += 1; } startupPass.processed += 1; @@ -237,8 +287,8 @@ export const make = (options: { readonly startupPause?: Duration.Input } = {}) = } }); - // Items wait for activation. Live cascades and startup orphans share this - // one worker, so they never race each other on the same child. + // Items wait for activation. Live work and startup orphans share this one + // worker, so they never race each other on the same child. const worker = yield* makeDrainableWorker((request: CascadeRequest) => activation.pipe( Effect.andThen(processRequest(request)), @@ -253,9 +303,9 @@ export const make = (options: { readonly startupPause?: Duration.Input } = {}) = ), ); - /** Queues the orphans one per pause, so live cascades run between them. */ + /** Queues the orphans one per pause, so live work runs between them. */ const queueStartupPass = Effect.gen(function* () { - const orphans = yield* selectOrphans(); + const orphans = yield* selectChildren("archive", { kind: "all" }); startupPass.candidates = orphans.length; for (const [index, orphan] of orphans.entries()) { if (index > 0 && !Duration.isZero(startupPause)) yield* Effect.sleep(startupPause); @@ -300,7 +350,7 @@ export const make = (options: { readonly startupPause?: Duration.Input } = {}) = return { start, - /** After `start`: waits for the whole startup pass, then for every queued cascade. */ + /** After `start`: waits for the whole startup pass, then for every queued item. */ drain: Deferred.await(startupQueued).pipe(Effect.andThen(worker.drain)), }; });