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..d44943873ab82 --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.test.ts @@ -0,0 +1,529 @@ +// T3-CUSTOM(expbkt3): provider-native subagents follow their parent into and out of the archive. +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + NodeId, + type OrchestrationV2AppThread, + type OrchestrationV2StoredEvent, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + 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"; +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 driver = ProviderDriverKind.make("codex"); + +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, + })), + }); + }); + +/** + * 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 parentThreadId = child.lineage.parentThreadId!; + const subagentId = NodeId.make(`subagent:${child.id}`); + yield* eventSink.write({ + events: [ + { + id: EventId.make(`event:${subagentId}:${status}`), + type: "subagent.updated", + threadId: parentThreadId, + nodeId: subagentId, + driver, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + id: subagentId, + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make(`node:${parentThreadId}:root`), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId: modelSelection.instanceId, + providerThreadId: null, + childThreadId: child.id, + nativeTaskRef: null, + prompt: "Explore the repository", + title: null, + model: null, + status, + result: status === "completed" ? "Done." : null, + startedAt: now, + completedAt: status === "completed" ? now : null, + updatedAt: now, + }, + }, + ], + }); + }); + +/** 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; + }); + +/** 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", + () => + 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({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + 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({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + 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({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + 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* seedSubagentRecord(stillRunning, "running", now); + + const cascade = yield* SubagentArchiveCascade.make({ + startupPause: Duration.zero, + cascadePause: Duration.zero, + }); + 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)), +); + +it.effect("the startup pass archives one orphan per pause", () => + Effect.scoped( + Effect.gen(function* () { + 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), + cascadePause: Duration.zero, + }); + 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 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)), +); + +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 new file mode 100644 index 0000000000000..a1af2fb4be24d --- /dev/null +++ b/apps/server/src/orchestration-v2/SubagentArchiveCascade.expbkt3.ts @@ -0,0 +1,359 @@ +/** + * 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. + * - 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. + * + * 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, + 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 Clock from "effect/Clock"; +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"; +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"; + +/** ~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 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" && + 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 = ( + 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; + + /** + * 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 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${ + 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') ${ + 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 ${ + 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 + `; + + /** 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}`, + ); + const command: OrchestrationV2ServerCommand = + direction === "archive" + ? { 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) => + 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)), + ), + ); + + const processRequest = Effect.fn("SubagentArchiveCascade.process")(function* ( + request: CascadeRequest, + ) { + switch (request.type) { + case "follow": { + const children = yield* selectChildren(request.direction, { + kind: "parent", + parentThreadId: request.parentThreadId, + projectId: request.projectId, + }); + if (children.length === 0) return; + 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, + }); + 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": { + if (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. 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)), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("subagent archive cascade: request failed", { + request, + cause: Cause.pretty(cause), + }), + ), + ), + ); + + /** Queues the orphans one per pause, so live work runs between them. */ + const queueStartupPass = Effect.gen(function* () { + 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); + 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 item. */ + 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())); 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),