From 6e7a6c54b216010168b42f070f891f66d22de017 Mon Sep 17 00:00:00 2001 From: nickleuze Date: Sun, 11 Oct 2026 01:08:07 +0200 Subject: [PATCH 1/2] fix(desktop): slim the shutdown watchdog to one hook Keep the 34-line watchdog, forked once after the shutdown request, and drop the scaffolding around it: the traceTeardown wraps around four upstream startup steps, the before-quit/signal breadcrumbs, the desktop-shutdown.log writer and the DesktopLifecycle test that only exercised unchanged upstream code. The watchdog now logs through the desktop logger and exits 1 when it has to force the quit, so a supervisor can see it. Upstream #17386 makes the backend exit while the desktop holds its pipes (the ~2s SIGKILL-after-grace in every recorded quit), but it does not bound desktop teardown, which is what the watchdog covers. Co-Authored-By: Claude Opus 5.5 --- apps/desktop/src/app/DesktopApp.ts | 26 ++--- apps/desktop/src/app/DesktopLifecycle.test.ts | 72 -------------- apps/desktop/src/app/DesktopLifecycle.ts | 3 - apps/desktop/src/app/DesktopShutdownLog.ts | 42 -------- .../src/app/DesktopShutdownWatchdog.test.ts | 4 +- .../src/app/DesktopShutdownWatchdog.ts | 17 ++-- .../src/mcp/toolkits/project/handlers.test.ts | 98 ------------------- 7 files changed, 20 insertions(+), 242 deletions(-) delete mode 100644 apps/desktop/src/app/DesktopShutdownLog.ts diff --git a/apps/desktop/src/app/DesktopApp.ts b/apps/desktop/src/app/DesktopApp.ts index 40dbc4705d8cd..426f070baef8b 100644 --- a/apps/desktop/src/app/DesktopApp.ts +++ b/apps/desktop/src/app/DesktopApp.ts @@ -26,7 +26,6 @@ import * as DesktopLinuxUrlHandler from "./DesktopLinuxUrlHandler.ts"; import * as DesktopObservability from "./DesktopObservability.ts"; import * as DesktopPreReadyPlatform from "./DesktopPreReadyPlatform.ts"; import * as DesktopShutdown from "./DesktopShutdown.ts"; -import { configureShutdownLog, shutdownBreadcrumb, traceTeardown } from "./DesktopShutdownLog.ts"; import { watchShutdown } from "./DesktopShutdownWatchdog.ts"; import * as DesktopServerExposure from "../backend/DesktopServerExposure.ts"; import * as DesktopAppSettings from "../settings/DesktopAppSettings.ts"; @@ -321,9 +320,8 @@ const startup = Effect.gen(function* () { } yield* appIdentity.configure; - configureShutdownLog(environment.logDir); - yield* traceTeardown("lifecycle listeners", lifecycle.register); - yield* traceTeardown("clerk", clerk.configure); + yield* lifecycle.register; + yield* clerk.configure; yield* electronApp.whenReady.pipe( Effect.withSpan("desktop.electron.whenReady"), @@ -339,13 +337,10 @@ const startup = Effect.gen(function* () { yield* appIdentity.configure; yield* previewPasskeys.configure; yield* applicationMenu.configure; - yield* traceTeardown("desktop updates", updates.configure); - yield* traceTeardown("remote updates", DesktopRemoteUpdates.listen); + yield* updates.configure; + yield* DesktopRemoteUpdates.listen; yield* linuxUrlHandler.register; - yield* traceTeardown( - "bootstrap", - bootstrap.pipe(Effect.catchCause((cause) => fatalStartupCause("bootstrap", cause))), - ); + yield* bootstrap.pipe(Effect.catchCause((cause) => fatalStartupCause("bootstrap", cause))); }).pipe(Effect.withSpan("desktop.startup")); const scopedProgram = Effect.scoped( @@ -363,20 +358,15 @@ const scopedProgram = Effect.scoped( // cascade, so leaving the WSL instance for its parent scope // finalizer means it gets hard-killed by the OS instead of // receiving SIGTERM + grace. - Effect.sync(() => shutdownBreadcrumb("stopping backends")).pipe( - Effect.andThen(stopAllPoolInstances()), + stopAllPoolInstances().pipe( Effect.ensuring(rendererHistory.shutdown), - Effect.ensuring( - Effect.sync(() => shutdownBreadcrumb("backends stopped; shutdown complete")).pipe( - Effect.andThen(shutdown.markComplete), - ), - ), + Effect.ensuring(shutdown.markComplete), ), ); yield* startup; yield* shutdown.awaitRequest; - shutdownBreadcrumb("shutdown requested; releasing app resources"); + // Scope teardown can stall; the watchdog bounds it. yield* watchShutdown(stopAllPoolInstances).pipe(Effect.forkDetach); }), ); diff --git a/apps/desktop/src/app/DesktopLifecycle.test.ts b/apps/desktop/src/app/DesktopLifecycle.test.ts index c98638f783e29..5caa18a53aabf 100644 --- a/apps/desktop/src/app/DesktopLifecycle.test.ts +++ b/apps/desktop/src/app/DesktopLifecycle.test.ts @@ -18,7 +18,6 @@ import * as DesktopWindow from "../window/DesktopWindow.ts"; function layerElectronApp( appListeners: Map void>, quit: Effect.Effect = Effect.void, - overrides: Partial = {}, ) { const registerListener = (eventName: string, listener: (...args: readonly unknown[]) => void) => Effect.acquireRelease( @@ -52,7 +51,6 @@ function layerElectronApp( onBeforeQuitForUpdate: (listener) => registerListener("before-quit-for-update", listener), on: (eventName, listener) => registerListener(eventName, listener as unknown as (...args: readonly unknown[]) => void), - ...overrides, } satisfies ElectronApp.ElectronApp["Service"]); } @@ -104,76 +102,6 @@ function layerDesktopWindow( } describe("DesktopLifecycle", () => { - it.effect.each([false, true])( - "waits for backend shutdown before relaunch (development=%s)", - (isDevelopment) => - Effect.gen(function* () { - const events: string[] = []; - const appListeners = new Map void>(); - const shutdownRequested = yield* Deferred.make(); - const shutdownComplete = yield* Deferred.make(); - const exited = yield* Deferred.make(); - const layer = DesktopLifecycle.layer.pipe( - Layer.provideMerge(layerElectronTheme), - Layer.provideMerge( - layerElectronApp(appListeners, Effect.void, { - relaunch: () => - Effect.sync(() => { - events.push("relaunch"); - }), - exit: (code) => - Effect.sync(() => { - events.push(`exit:${code}`); - }).pipe(Effect.andThen(Deferred.succeed(exited, code ?? 0)), Effect.asVoid), - }), - ), - Layer.provideMerge( - layerDesktopWindow({ - flushMainWindowBounds: Effect.sync(() => { - events.push("flush"); - }), - }), - ), - Layer.provideMerge( - Layer.succeed(DesktopEnvironment.DesktopEnvironment, { - platform: "darwin", - isDevelopment, - } as DesktopEnvironment.DesktopEnvironment["Service"]), - ), - Layer.provideMerge( - Layer.succeed(DesktopShutdown.DesktopShutdown, { - request: Effect.sync(() => { - events.push("shutdown"); - }).pipe( - Effect.andThen(Deferred.succeed(shutdownRequested, undefined)), - Effect.asVoid, - ), - awaitRequest: Deferred.await(shutdownRequested), - awaitComplete: Deferred.await(shutdownComplete), - markComplete: Deferred.succeed(shutdownComplete, undefined).pipe(Effect.asVoid), - isComplete: Deferred.isDone(shutdownComplete), - }), - ), - Layer.provideMerge(DesktopState.layer), - ); - yield* Effect.gen(function* () { - const lifecycle = yield* DesktopLifecycle.DesktopLifecycle; - yield* lifecycle.relaunch("test-only fake relaunch"); - yield* Deferred.await(shutdownRequested); - assert.deepEqual(events, ["flush", "shutdown"]); - assert.isFalse(yield* Deferred.isDone(exited)); - yield* Deferred.succeed(shutdownComplete, undefined); - assert.equal(yield* Deferred.await(exited), isDevelopment ? 75 : 0); - assert.deepEqual( - events, - isDevelopment - ? ["flush", "shutdown", "exit:75"] - : ["flush", "shutdown", "relaunch", "exit:0"], - ); - }).pipe(Effect.provide(layer)); - }), - ); - it.effect.each(["darwin", "win32", "linux"] satisfies ReadonlyArray)( "lets the updater's quit event proceed on %s", (platform) => { diff --git a/apps/desktop/src/app/DesktopLifecycle.ts b/apps/desktop/src/app/DesktopLifecycle.ts index bc72699b19744..924341781aa6b 100644 --- a/apps/desktop/src/app/DesktopLifecycle.ts +++ b/apps/desktop/src/app/DesktopLifecycle.ts @@ -10,7 +10,6 @@ import type * as Electron from "electron"; import * as DesktopEnvironment from "./DesktopEnvironment.ts"; import { makeComponentLogger } from "./DesktopObservability.ts"; import * as DesktopShutdown from "./DesktopShutdown.ts"; -import { shutdownBreadcrumb } from "./DesktopShutdownLog.ts"; import * as ElectronApp from "../electron/ElectronApp.ts"; import * as ElectronTheme from "../electron/ElectronTheme.ts"; import * as ElectronWindow from "../electron/ElectronWindow.ts"; @@ -103,7 +102,6 @@ function handleBeforeQuit( allowQuit: () => boolean, markQuitAllowed: () => void, ): void { - shutdownBreadcrumb(`before-quit (${allowQuit() ? "quit allowed" : "shutting down first"})`); if (allowQuit()) { void runEffect( Effect.gen(function* () { @@ -153,7 +151,6 @@ function quitFromSignal( const electronApp = yield* ElectronApp.ElectronApp; const state = yield* DesktopState.DesktopState; const wasQuitting = yield* Ref.getAndSet(state.quitting, true); - shutdownBreadcrumb(`${signal} received${wasQuitting ? " while already quitting" : ""}`); if (wasQuitting) return; yield* logLifecycleInfo("process signal received", { signal }); yield* requestDesktopShutdownAndWait(); diff --git a/apps/desktop/src/app/DesktopShutdownLog.ts b/apps/desktop/src/app/DesktopShutdownLog.ts deleted file mode 100644 index c8299c3edb2a3..0000000000000 --- a/apps/desktop/src/app/DesktopShutdownLog.ts +++ /dev/null @@ -1,42 +0,0 @@ -// @effect-diagnostics nodeBuiltinImport:off -- Breadcrumbs must reach disk synchronously to survive a stalled or killed quit. -// @effect-diagnostics globalDate:off -- Breadcrumbs are written outside any Effect fiber, from signal and quit handlers. -import * as NodeFS from "node:fs"; -import * as NodePath from "node:path"; - -import * as Effect from "effect/Effect"; -import type * as Scope from "effect/Scope"; - -/** - * Shutdown breadcrumbs, appended synchronously to `desktop-shutdown.log` so - * they survive a stalled or killed quit. Spans only reach the trace when they - * end, which a hung shutdown never does. - */ -let shutdownLogPath: string | null = null; - -export function configureShutdownLog(logDir: string): void { - shutdownLogPath = NodePath.join(logDir, "desktop-shutdown.log"); -} - -export function shutdownBreadcrumb(message: string): void { - if (shutdownLogPath === null) return; - try { - NodeFS.appendFileSync(shutdownLogPath, `${new Date().toISOString()} ${message}\n`); - } catch { - // A missing log is no reason to disturb the quit. - } -} - -/** - * Runs `effect` with breadcrumbs around the teardown of everything it adds to - * the current scope: "releasing" when that teardown starts, "released" once it - * has finished. A label with no "released" line names the stalled resource. - */ -export const traceTeardown = Effect.fnUntraced(function* ( - label: string, - effect: Effect.Effect, -): Effect.fn.Return { - yield* Effect.addFinalizer(() => Effect.sync(() => shutdownBreadcrumb(`${label}: released`))); - const result = yield* effect; - yield* Effect.addFinalizer(() => Effect.sync(() => shutdownBreadcrumb(`${label}: releasing`))); - return result; -}); diff --git a/apps/desktop/src/app/DesktopShutdownWatchdog.test.ts b/apps/desktop/src/app/DesktopShutdownWatchdog.test.ts index 6a002a4a89631..01ed9561011d6 100644 --- a/apps/desktop/src/app/DesktopShutdownWatchdog.test.ts +++ b/apps/desktop/src/app/DesktopShutdownWatchdog.test.ts @@ -28,7 +28,7 @@ it.effect("stalled resource cleanup stops backends and allows the quit to finish yield* Deferred.await(stopped); yield* shutdown.awaitComplete; yield* TestClock.adjust("10 seconds"); - assert.strictEqual(yield* Deferred.await(exited), 0); + assert.strictEqual(yield* Deferred.await(exited), 1); yield* Fiber.join(fiber); }).pipe(Effect.provide(DesktopShutdown.layer)), ); @@ -45,7 +45,7 @@ it.effect("a stalled emergency backend stop cannot stall the independent exit de Effect.forkChild, ); yield* TestClock.adjust("30 seconds"); - assert.strictEqual(yield* Deferred.await(exited), 0); + assert.strictEqual(yield* Deferred.await(exited), 1); yield* Fiber.join(fiber); }).pipe(Effect.provide(DesktopShutdown.layer)), ); diff --git a/apps/desktop/src/app/DesktopShutdownWatchdog.ts b/apps/desktop/src/app/DesktopShutdownWatchdog.ts index 260641e6675a7..d15b5f33a8c8a 100644 --- a/apps/desktop/src/app/DesktopShutdownWatchdog.ts +++ b/apps/desktop/src/app/DesktopShutdownWatchdog.ts @@ -2,12 +2,15 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as DesktopShutdown from "./DesktopShutdown.ts"; -import { shutdownBreadcrumb } from "./DesktopShutdownLog.ts"; +import { makeComponentLogger } from "./DesktopObservability.ts"; import * as ElectronApp from "../electron/ElectronApp.ts"; +const { logWarning } = makeComponentLogger("desktop-shutdown-watchdog"); + /** - * Teardown can stall before its finalizer stops the backends. The independent - * exit deadline must also survive a stalled emergency backend stop. + * Bounds a requested shutdown. Teardown can stall before its finalizer stops + * the backends, so after 20s the watchdog stops them itself and completes the + * shutdown; an independent 30s deadline exits even if that stop stalls too. * The callback belongs to the app, which owns the complete backend pool. */ export const watchShutdown = Effect.fn("desktop.app.shutdownWatchdog")(function* ( @@ -18,15 +21,15 @@ export const watchShutdown = Effect.fn("desktop.app.shutdownWatchdog")(function* const recover = Effect.gen(function* () { yield* Effect.sleep(Duration.seconds(20)); if (yield* shutdown.isComplete) return; - shutdownBreadcrumb("shutdown stalled; stopping backends directly"); + yield* logWarning("shutdown stalled; stopping backends directly"); yield* stopBackends(); - shutdownBreadcrumb("backends stopped by the watchdog"); yield* shutdown.markComplete; }); const exit = Effect.gen(function* () { yield* Effect.sleep(Duration.seconds(30)); - shutdownBreadcrumb("app did not exit after shutdown; exiting"); - yield* electronApp.exit(0); + yield* logWarning("app did not exit after shutdown; exiting"); + // Non-zero, so a supervisor sees that the quit had to be forced. + yield* electronApp.exit(1); }); // In the real app exit terminates the process. In tests it completes the // deadline and interrupts any still-stalled recovery fiber. diff --git a/apps/server/src/mcp/toolkits/project/handlers.test.ts b/apps/server/src/mcp/toolkits/project/handlers.test.ts index 61fa34664dd24..b74df9dae1f99 100644 --- a/apps/server/src/mcp/toolkits/project/handlers.test.ts +++ b/apps/server/src/mcp/toolkits/project/handlers.test.ts @@ -99,104 +99,6 @@ it.effect("attributes a launched thread's first message to the calling thread", }), ); -it.effect( - "an auto caller launches with inherited workspace and sender, without escalating modes", - () => - Effect.gen(function* () { - const sourceThreadId = ThreadId.make("source-thread"); - const projectId = ProjectId.make("project"); - const providerInstanceId = ProviderInstanceId.make("codex"); - const modelSelection = { instanceId: providerInstanceId, model: "gpt-5" }; - let caller = { - id: sourceThreadId, - projectId, - providerInstanceId, - modelSelection, - runtimeMode: "auto", - interactionMode: "default", - activeRunId: "active-run", - archivedAt: null, - deletedAt: null, - } as OrchestrationV2ThreadShell; - const launched: Array = []; - const layerDependencies = Layer.mergeAll( - NodeCrypto.layer, - Layer.succeed(McpInvocationContext.McpInvocationContext, { - environmentId: EnvironmentId.make("environment"), - requestNamespace: "session", - thread: { - threadId: sourceThreadId, - providerSessionId: "session", - providerInstanceId, - }, - client: undefined, - issuedAt: 0, - capabilities: new Set(["orchestration" as const]), - }), - Layer.mock(ThreadManagement.ThreadManagementService)({ - getThreadShell: () => Effect.succeed(caller), - }), - Layer.mock(ThreadLaunch.ThreadLaunchService)({ - launch: (input) => { - launched.push(input); - return Effect.succeed({ - threadId: input.threadId, - projection: { - thread: { id: input.threadId, projectId, modelSelection }, - runs: [], - }, - resumed: false, - } as unknown as ThreadLaunch.ThreadLaunchResult); - }, - }), - Layer.mock(Project.ProjectService)({}), - Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/projects" }), - Layer.mock(GitVcsDriver.GitVcsDriver)({}), - NodeServices.layer, - ServerConfig.layerTest(process.cwd(), { prefix: "t3-source-link-" }).pipe( - Layer.provide(NodeServices.layer), - ), - ); - const toolkit = yield* ProjectToolkit.pipe( - Effect.provide( - McpToolAccess.HandlersLayer.layer(ProjectHandlers.layer).pipe( - Layer.provide(layerDependencies), - ), - ), - ); - const handle = (params: Parameters>[1]) => - toolkit - .handle("t3_thread_launch", params) - .pipe(Stream.unwrap, Stream.runCollect, Effect.provide(layerDependencies)); - expect((yield* handle({ title: "Inherited", message: "Go" })).at(-1)?.result).toMatchObject({ - projectId, - modelSelection, - }); - yield* handle({ title: "Narrower", message: "Go", runtimeMode: "approval-required" }); - expect(launched.map((input) => input.runtimeMode)).toEqual(["auto", "approval-required"]); - expect(launched.map((input) => input.workspaceStrategy)).toEqual([ - { type: "root" }, - { type: "root" }, - ]); - expect( - launched.every((input) => input.initialMessage?.senderThreadId === sourceThreadId), - ).toBe(true); - expect( - (yield* handle({ title: "Broader", runtimeMode: "full-access" })).at(-1)?.result, - ).toMatchObject({ code: "runtime_mode_escalation_denied" }); - caller = { ...caller, interactionMode: "plan" }; - expect( - (yield* handle({ title: "Broader interaction", interactionMode: "default" })).at(-1) - ?.result, - ).toMatchObject({ code: "interaction_mode_escalation_denied" }); - caller = { ...caller, activeRunId: null }; - expect((yield* handle({ title: "Inactive" })).at(-1)?.result).toMatchObject({ - code: "parent_not_active", - }); - expect(launched).toHaveLength(2); - }), -); - it.effect("launches a scratch thread into the Scratch project", () => Effect.gen(function* () { const sourceThreadId = ThreadId.make("source-thread"); From 1a043f37dae65a2369b1206eb21a1018da0225a7 Mon Sep 17 00:00:00 2001 From: nickleuze Date: Sun, 11 Oct 2026 01:08:09 +0200 Subject: [PATCH 2/2] fix(server): launched threads report back across restarts ThreadReportBack now only reports finished runs. A startup sweep, run after recovery, re-offers runs from the last day that another thread's agent started, so runs that ended unreported while the server was going down, and runs restart reconciliation stopped, still reach the sender. The deterministic command ids keep delivery at most once. Removed: waiting-for-user notices and their timers, the test-only sequence-tracking drain (tests await the notice event and use the worker's drain), and the project handler test that re-tested upstream's launch gate. Agents are told launched/sent threads report back. Co-Authored-By: Claude Opus 5.5 --- .../orchestration-v2/ThreadReportBack.test.ts | 415 +++++++++--------- .../src/orchestration-v2/ThreadReportBack.ts | 337 ++++---------- docs/operations/personal-fork.md | 2 +- .../src/server/orchestrationInstructions.ts | 1 + 4 files changed, 309 insertions(+), 446 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadReportBack.test.ts b/apps/server/src/orchestration-v2/ThreadReportBack.test.ts index 6b6695c4339b2..998c062445ae0 100644 --- a/apps/server/src/orchestration-v2/ThreadReportBack.test.ts +++ b/apps/server/src/orchestration-v2/ThreadReportBack.test.ts @@ -10,19 +10,19 @@ import { NodeId, type OrchestrationV2ConversationMessage, type OrchestrationV2Run, - type OrchestrationV2RuntimeRequest, ProjectId, ProviderDriverKind, ProviderInstanceId, RunId, - RuntimeRequestId, ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; +import type * as SqlClient from "effect/sql/SqlClient"; import * as Stream from "effect/Stream"; -import * as TestClock from "effect/testing/TestClock"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as ServerConfig from "../config.ts"; @@ -79,61 +79,65 @@ const providerInstance = { textGeneration: {} as ProviderInstance["textGeneration"], } satisfies ProviderInstance; -const TestLayer = Layer.mergeAll(layer, layerEventSink, EventStore.layer).pipe( - Layer.provideMerge(layerProjectService), - Layer.provide( - Layer.mock(WorkspacePaths.WorkspacePaths)({ - normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot), - }), - ), - Layer.provide(ProviderTurnStartServiceTestkit.layer), - Layer.provide( - Layer.succeed(ProjectEnrichmentService.ProjectEnrichmentService, { - peek: () => - Effect.succeed({ - repositoryIdentity: null, - faviconPath: null, - repositoryIdentityResolved: false, - }), - request: () => Effect.void, - getAvailable: () => - Effect.succeed({ - repositoryIdentity: null, - faviconPath: null, - repositoryIdentityResolved: false, - }), - invalidate: () => Effect.void, - subscribeChanges: Effect.never, - }), - ), - Layer.provide(McpSessionRegistryTestkit.layer), - Layer.provide(SqlitePersistence.layerMemory), - Layer.provide( - CheckpointStore.layer.pipe( - Layer.provide(VcsDriverRegistry.layer), - Layer.provide(VcsProcess.layer), - Layer.provide(ServerConfigLayer), - Layer.provide(PlatformTestLayer), +const makeTestLayer = (database: Layer.Layer) => + Layer.mergeAll(layer, layerEventSink, EventStore.layer).pipe( + Layer.provideMerge(layerProjectService), + Layer.provide( + Layer.mock(WorkspacePaths.WorkspacePaths)({ + normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot), + }), ), - ), - Layer.provide(ServerConfigLayer), - Layer.provide(ServerSettings.layerTest()), - Layer.provide( - Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { - getInstance: (instanceId) => - Effect.succeed(instanceId === providerInstance.instanceId ? providerInstance : undefined), - listInstances: Effect.succeed([providerInstance]), - listUnavailable: Effect.succeed([]), - streamChanges: Stream.empty, - subscribeChanges: Effect.never, - }), - ), - Layer.provide(PlatformTestLayer), - Layer.provide(McpProviderSessions.layer), -); + Layer.provide(ProviderTurnStartServiceTestkit.layer), + Layer.provide( + Layer.succeed(ProjectEnrichmentService.ProjectEnrichmentService, { + peek: () => + Effect.succeed({ + repositoryIdentity: null, + faviconPath: null, + repositoryIdentityResolved: false, + }), + request: () => Effect.void, + getAvailable: () => + Effect.succeed({ + repositoryIdentity: null, + faviconPath: null, + repositoryIdentityResolved: false, + }), + invalidate: () => Effect.void, + subscribeChanges: Effect.never, + }), + ), + Layer.provide(McpSessionRegistryTestkit.layer), + Layer.provideMerge(database), + Layer.provide( + CheckpointStore.layer.pipe( + Layer.provide(VcsDriverRegistry.layer), + Layer.provide(VcsProcess.layer), + Layer.provide(ServerConfigLayer), + Layer.provide(PlatformTestLayer), + ), + ), + Layer.provide(ServerConfigLayer), + Layer.provide(ServerSettings.layerTest()), + Layer.provide( + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, { + getInstance: (instanceId) => + Effect.succeed(instanceId === providerInstance.instanceId ? providerInstance : undefined), + listInstances: Effect.succeed([providerInstance]), + listUnavailable: Effect.succeed([]), + streamChanges: Stream.empty, + subscribeChanges: Effect.never, + }), + ), + Layer.provide(PlatformTestLayer), + Layer.provide(McpProviderSessions.layer), + ); -/** A project with a sending thread S and a worker thread X, plus a running reactor. */ -const setup = Effect.fn("ThreadReportBackTest.setup")(function* (name: string) { +/** A project with a sending thread S and worker threads, without a running reactor. */ +const setupThreads = Effect.fn("ThreadReportBackTest.setupThreads")(function* ( + name: string, + workers: ReadonlyArray = ["Helper"], +) { const projects = yield* ProjectService.ProjectService; const orchestrator = yield* Orchestrator.OrchestratorV2; const projectId = ProjectId.make(`project:${name}`); @@ -159,12 +163,22 @@ const setup = Effect.fn("ThreadReportBackTest.setup")(function* (name: string) { worktreePath: null, }); const sender = ThreadId.make(`thread:${name}:sender`); - const worker = ThreadId.make(`thread:${name}:worker`); yield* createThread(sender, "Orchestrator"); - yield* createThread(worker, "Helper"); + const workerIds: Array = []; + for (const [index, title] of workers.entries()) { + const worker = ThreadId.make(`thread:${name}:worker-${index}`); + yield* createThread(worker, title); + workerIds.push(worker); + } + return { sender, workers: workerIds, createThread }; +}); + +/** A sending thread S and a worker thread X, plus a running reactor. */ +const setup = Effect.fn("ThreadReportBackTest.setup")(function* (name: string) { + const threads = yield* setupThreads(name); const reactor = yield* ThreadReportBack.make; yield* reactor.start(); - return { sender, worker, reactor, createThread }; + return { ...threads, worker: threads.workers[0]!, reactor }; }); let eventCounter = 0; @@ -260,10 +274,12 @@ const finishRun = Effect.fn("ThreadReportBackTest.finishRun")(function* ( threadId: ThreadId, runId: RunId, status: OrchestrationV2Run["status"], + commandId?: CommandId, ) { const sink = yield* EventSink.EventSinkV2; const now = yield* DateTime.now; yield* sink.write({ + ...(commandId === undefined ? {} : { commandId }), events: [ { id: nextEventId(`${runId}:${status}`), @@ -277,39 +293,6 @@ const finishRun = Effect.fn("ThreadReportBackTest.finishRun")(function* ( }); }); -const writeRequest = Effect.fn("ThreadReportBackTest.writeRequest")(function* ( - threadId: ThreadId, - runId: RunId, - requestId: RuntimeRequestId, - status: OrchestrationV2RuntimeRequest["status"], - kind: OrchestrationV2RuntimeRequest["kind"] = "command", -) { - const sink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - yield* sink.write({ - events: [ - { - id: nextEventId(`${requestId}:${status}`), - type: "runtime-request.updated", - threadId, - runId, - occurredAt: now, - payload: { - id: requestId, - nodeId: NodeId.make(`node:${requestId}`), - providerTurnId: null, - nativeRequestRef: null, - kind, - status, - responseCapability: { type: "message" }, - createdAt: now, - resolvedAt: status === "pending" ? null : now, - }, - }, - ], - }); -}); - const notices = (threadId: ThreadId) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -319,20 +302,43 @@ const notices = (threadId: ThreadId) => return messages.filter((message) => message.notification !== undefined); }); +/** + * Runs `trigger` and waits for the notice from `source` it causes `recipient` + * to receive. The reactor handles runs in commit order, so runs that ended + * before the trigger's run have been decided by then. + */ +const awaitNoticeAfter = ( + recipient: ThreadId, + source: ThreadId, + trigger: Effect.Effect, +) => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const afterSequence = yield* sink.latestSequence(); + yield* trigger; + const notice = yield* sink.stream({ afterSequence, eventType: "message.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "message.updated" && + stored.event.threadId === recipient && + stored.event.payload.notification !== undefined && + stored.event.payload.senderThreadId === source, + ), + Stream.runHead, + ); + assert.isTrue(notice._tag === "Some"); + }); + const fromAgent = (sender: ThreadId) => ({ createdBy: "agent", creationSource: "mcp", senderThreadId: sender }) as const; -it.layer(TestLayer)("ThreadReportBack", (it) => { +it.layer(makeTestLayer(SqlitePersistence.layerMemory))("ThreadReportBack", (it) => { it.effect("tells the launching thread when its launched thread finishes", () => Effect.gen(function* () { - const { sender, worker, reactor } = yield* setup("launched"); + const { sender, worker, createThread } = yield* setup("launched"); const runId = RunId.make("run:launched"); yield* writeRunningTurn({ threadId: worker, runId, starter: fromAgent(sender) }); - yield* reactor.drain; - assert.lengthOf(yield* notices(sender), 0); - - yield* finishRun(worker, runId, "completed"); - yield* reactor.drain; + yield* awaitNoticeAfter(sender, worker, finishRun(worker, runId, "completed")); const [notice, ...rest] = yield* notices(sender); assert.lengthOf(rest, 0); @@ -351,14 +357,18 @@ it.layer(TestLayer)("ThreadReportBack", (it) => { // A replayed terminal update reuses the same command id, so it cannot deliver twice. yield* finishRun(worker, runId, "completed"); - yield* reactor.drain; - assert.lengthOf(yield* notices(sender), 1); + const later = ThreadId.make("thread:launched:later"); + yield* createThread(later, "Later helper"); + const laterRun = RunId.make("run:launched-later"); + yield* writeRunningTurn({ threadId: later, runId: laterRun, starter: fromAgent(sender) }); + yield* awaitNoticeAfter(sender, later, finishRun(later, laterRun, "completed")); + assert.lengthOf(yield* notices(sender), 2); }).pipe(Effect.scoped), ); it.effect("tells a thread that steered a turn with t3_thread_send", () => Effect.gen(function* () { - const { sender, worker, reactor } = yield* setup("steered"); + const { sender, worker } = yield* setup("steered"); const runId = RunId.make("run:steered"); yield* writeRunningTurn({ threadId: worker, @@ -366,15 +376,14 @@ it.layer(TestLayer)("ThreadReportBack", (it) => { starter: {}, steered: [fromAgent(sender)], }); - yield* finishRun(worker, runId, "completed"); - yield* reactor.drain; + yield* awaitNoticeAfter(sender, worker, finishRun(worker, runId, "completed")); assert.lengthOf(yield* notices(sender), 1); }).pipe(Effect.scoped), ); it.effect("reports failed and stopped turns with their outcome", () => Effect.gen(function* () { - const { sender, worker, reactor, createThread } = yield* setup("outcomes"); + const { sender, worker, createThread } = yield* setup("outcomes"); const failedRun = RunId.make("run:outcomes-failed"); yield* writeRunningTurn({ threadId: worker, runId: failedRun, starter: fromAgent(sender) }); yield* finishRun(worker, failedRun, "failed"); @@ -383,8 +392,7 @@ it.layer(TestLayer)("ThreadReportBack", (it) => { yield* createThread(stopped, "Stopped helper"); const stoppedRun = RunId.make("run:outcomes-stopped"); yield* writeRunningTurn({ threadId: stopped, runId: stoppedRun, starter: fromAgent(sender) }); - yield* finishRun(stopped, stoppedRun, "interrupted"); - yield* reactor.drain; + yield* awaitNoticeAfter(sender, stopped, finishRun(stopped, stoppedRun, "interrupted")); const received = (yield* notices(sender)).map((message) => message.notification); assert.sameDeepMembers(received, [ @@ -403,12 +411,12 @@ it.layer(TestLayer)("ThreadReportBack", (it) => { ); it.effect( - "stays quiet for user turns, notice turns, delegated children and archived senders", + "stays quiet for user turns, notice turns, delegated children, archived senders and restart reconciliation", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; - const { sender, worker, reactor, createThread } = yield* setup("quiet"); + const { sender, worker, createThread } = yield* setup("quiet"); const userRun = RunId.make("run:quiet-user"); yield* writeRunningTurn({ threadId: worker, runId: userRun, starter: {} }); @@ -460,8 +468,20 @@ it.layer(TestLayer)("ThreadReportBack", (it) => { const childRun = RunId.make("run:quiet-child"); yield* writeRunningTurn({ threadId: child, runId: childRun, starter: fromAgent(sender) }); yield* finishRun(child, childRun, "completed"); - yield* reactor.drain; - assert.lengthOf(yield* notices(sender), 0); + + // Live reconciliation is left to the startup sweep. + const reconciledRun = RunId.make("run:quiet-reconciled"); + yield* writeRunningTurn({ + threadId: worker, + runId: reconciledRun, + starter: fromAgent(sender), + }); + yield* finishRun( + worker, + reconciledRun, + "cancelled", + CommandId.make("command:runtime-reconcile:shutdown:quiet"), + ); // An archived sender is not woken. const archived = ThreadId.make("thread:quiet:archived-sender"); @@ -476,102 +496,93 @@ it.layer(TestLayer)("ThreadReportBack", (it) => { threadId: worker, runId: archivedRun, starter: fromAgent(archived), + steered: [fromAgent(sender)], }); - yield* finishRun(worker, archivedRun, "completed"); - yield* reactor.drain; + + // The last run reports, so every earlier one has been decided. + yield* awaitNoticeAfter(sender, worker, finishRun(worker, archivedRun, "completed")); + const received = yield* notices(sender); + assert.deepEqual( + received.map((message) => message.id), + [MessageId.make(`thread-report-back:finished:${worker}:${archivedRun}:${sender}`)], + ); assert.lengthOf(yield* notices(archived), 0); }).pipe(Effect.scoped), ); +}); - it.effect("tells the sender once when a turn keeps waiting on the user", () => - Effect.gen(function* () { - const { sender, worker, reactor } = yield* setup("waiting"); - const runId = RunId.make("run:waiting"); - const requestId = RuntimeRequestId.make("request:waiting"); - yield* writeRunningTurn({ threadId: worker, runId, starter: fromAgent(sender) }); - yield* writeRequest(worker, runId, requestId, "pending"); - yield* reactor.drain; - yield* TestClock.adjust(ThreadReportBack.WAITING_NOTICE_DELAY_MS - 1); - yield* reactor.drain; - assert.lengthOf(yield* notices(sender), 0); - - yield* TestClock.adjust(1); - yield* reactor.drain; - const [notice] = yield* notices(sender); - assert.deepEqual(notice?.notification, { - source: { kind: "subagent", childThreadId: worker }, - outcome: "updated", - summary: 'Thread "Helper" is waiting for the user', +it.effect("delivers reports a restart left pending, once, after reopening the database", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-thread-report-back-" }); + // Each runtime opens and closes its own connection to the same file. + const runtime = () => + makeTestLayer(SqlitePersistence.layerFromPath(path.join(directory, "state.sqlite"))); + const delivered = RunId.make("run:restart-delivered"); + const unreported = RunId.make("run:restart-unreported"); + const reconciled = RunId.make("run:restart-reconciled"); + + const { sender, workers } = yield* Effect.gen(function* () { + const threads = yield* setupThreads("restart", ["Delivered", "Unreported", "Reconciled"]); + const [deliveredThread, unreportedThread, reconciledThread] = threads.workers; + yield* writeRunningTurn({ + threadId: deliveredThread!, + runId: delivered, + starter: fromAgent(threads.sender), }); - assert.include(notice?.text ?? "", "waiting for the user to approve a command"); - assert.include(notice?.text ?? "", "unless the user told you to"); - - // A repeated pending update for the same request re-arms, but delivers nothing new. - yield* writeRequest(worker, runId, requestId, "pending"); - yield* reactor.drain; - yield* TestClock.adjust(ThreadReportBack.WAITING_NOTICE_DELAY_MS); - yield* reactor.drain; - assert.lengthOf(yield* notices(sender), 1); - }).pipe(Effect.scoped), - ); - - it.effect("sends nothing when the request resolves before the delay", () => - Effect.gen(function* () { - const { sender, worker, reactor } = yield* setup("resolved"); - const runId = RunId.make("run:resolved"); - const requestId = RuntimeRequestId.make("request:resolved"); - yield* writeRunningTurn({ threadId: worker, runId, starter: fromAgent(sender) }); - yield* writeRequest(worker, runId, requestId, "pending", "user_input"); - yield* reactor.drain; - yield* writeRequest(worker, runId, requestId, "resolved", "user_input"); - yield* reactor.drain; - yield* TestClock.adjust(ThreadReportBack.WAITING_NOTICE_DELAY_MS); - yield* reactor.drain; - assert.lengthOf(yield* notices(sender), 0); - }).pipe(Effect.scoped), - ); - - it.effect("tells the parent when a delegated child waits on the user", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const { sender: parent, worker, reactor } = yield* setup("delegated-waiting"); - const child = ThreadId.make("thread:delegated-waiting:child"); - const { thread: template } = yield* orchestrator.getThreadRecords(worker, []); - const now = yield* DateTime.now; - yield* sink.write({ - events: [ - { - id: EventId.make("event:delegated-waiting-child-created"), - type: "thread.created", - threadId: child, - occurredAt: now, - payload: { - ...template, - id: child, - title: "Delegated child", - lineage: { - parentThreadId: parent, - relationshipToParent: "subagent", - rootThreadId: parent, - }, - forkedFrom: { type: "node", nodeId: NodeId.make("node:delegated-waiting-task") }, - }, - }, - ], + yield* writeRunningTurn({ + threadId: unreportedThread!, + runId: unreported, + starter: fromAgent(threads.sender), }); - const runId = RunId.make("run:delegated-waiting"); - yield* writeRunningTurn({ threadId: child, runId, starter: {} }); - yield* writeRequest(child, runId, RuntimeRequestId.make("request:delegated"), "pending"); - yield* reactor.drain; - yield* TestClock.adjust(ThreadReportBack.WAITING_NOTICE_DELAY_MS); - yield* reactor.drain; - const [notice, ...rest] = yield* notices(parent); - assert.lengthOf(rest, 0); - assert.equal( - notice?.notification?.summary, - 'Thread "Delegated child" is waiting for the user', + yield* writeRunningTurn({ + threadId: reconciledThread!, + runId: reconciled, + starter: fromAgent(threads.sender), + }); + // Reported live before the restart. + yield* Effect.scoped( + Effect.gen(function* () { + const reactor = yield* ThreadReportBack.make; + yield* reactor.start(); + yield* awaitNoticeAfter( + threads.sender, + deliveredThread!, + finishRun(deliveredThread!, delivered, "completed"), + ); + }), ); - }).pipe(Effect.scoped), - ); -}); + // Ends after the reactor stopped, as when the server goes down before delivering. + yield* finishRun(unreportedThread!, unreported, "completed"); + assert.lengthOf(yield* notices(threads.sender), 1); + return threads; + }).pipe(Effect.provide(runtime())); + + const afterRestart = Effect.gen(function* () { + const reactor = yield* ThreadReportBack.make; + yield* reactor.start(); + // Startup reconciliation stops the turn that was still running. + yield* finishRun( + workers[2]!, + reconciled, + "cancelled", + CommandId.make("command:runtime-reconcile:startup:restart"), + ); + yield* reactor.sweep; + yield* reactor.drain; + return (yield* notices(sender)).map((message) => message.id); + }).pipe(Effect.scoped); + + const expected = [ + [workers[0]!, delivered], + [workers[1]!, unreported], + [workers[2]!, reconciled], + ].map(([thread, run]) => + MessageId.make(`thread-report-back:finished:${thread}:${run}:${sender}`), + ); + assert.sameMembers(yield* afterRestart.pipe(Effect.provide(runtime())), expected); + // A second restart finds nothing new to deliver. + assert.sameMembers(yield* afterRestart.pipe(Effect.provide(runtime())), expected); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); diff --git a/apps/server/src/orchestration-v2/ThreadReportBack.ts b/apps/server/src/orchestration-v2/ThreadReportBack.ts index b89bef6038ca0..c94363b0919d1 100644 --- a/apps/server/src/orchestration-v2/ThreadReportBack.ts +++ b/apps/server/src/orchestration-v2/ThreadReportBack.ts @@ -5,43 +5,28 @@ import { type OrchestrationV2ConversationMessage, type OrchestrationV2Notification, type OrchestrationV2Run, - type OrchestrationV2RuntimeRequest, type OrchestrationV2StoredEvent, - type RunId, - type RuntimeRequestId, type ThreadId, } from "@t3tools/contracts"; import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import * as Cause from "effect/Cause"; -import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; -import * as FiberMap from "effect/FiberMap"; import * as Layer from "effect/Layer"; -import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; -import * as TxRef from "effect/TxRef"; +import * as SqlClient from "effect/sql/SqlClient"; +import { forkParked } from "../serverActivation.ts"; import * as EventSink from "./EventSink.ts"; -import * as EventStore from "./EventStore.ts"; import * as Orchestrator from "./Orchestrator.ts"; -/** How long a run may wait on the user before the thread that started it hears about it. */ -export const WAITING_NOTICE_DELAY_MS = 20_000; - const PREAMBLE = "Automatic update from T3 Code (not a message from the user, and not approval):"; const TITLE_MAX_LENGTH = 80; -/** The only event types the reactor subscribes to, so it never retains unrelated tool bodies. */ -const OBSERVED_TYPES = ["run.updated", "runtime-request.updated"] as const; +/** How far back the startup sweep looks for runs whose report was never delivered. */ +const SWEEP_LOOKBACK = Duration.days(1); -type Work = - | { readonly type: "finished"; readonly run: OrchestrationV2Run } - | { - readonly type: "waiting"; - readonly threadId: ThreadId; - readonly requestId: RuntimeRequestId; - readonly runId: RunId | undefined; - }; +type FinishedRun = Pick; const FINISHED: Partial< Record< @@ -56,15 +41,6 @@ const FINISHED: Partial< rolled_back: { outcome: "cancelled", verb: "was rolled back" }, }; -const WAITING_FOR: Partial> = { - user_input: "is waiting for the user to answer a question", - "mcp-elicitation": "is waiting for the user to respond to an MCP request", - command: "is waiting for the user to approve a command", - "file-read": "is waiting for the user to approve a file read", - "file-change": "is waiting for the user to approve a file change", - permission: "is waiting for the user to grant a permission", -}; - function titleOf(thread: OrchestrationV2AppThread): string { const firstLine = thread.title.trim().split("\n")[0]?.trim() ?? ""; return firstLine.length > TITLE_MAX_LENGTH @@ -73,12 +49,12 @@ function titleOf(thread: OrchestrationV2AppThread): string { } /** App-owned delegate_task children already report to their parent through the delegated-task path. */ -function appOwnedSubagentParent(thread: OrchestrationV2AppThread): ThreadId | undefined { - return thread.lineage.relationshipToParent === "subagent" && +function isAppOwnedSubagent(thread: OrchestrationV2AppThread): boolean { + return ( + thread.lineage.relationshipToParent === "subagent" && thread.lineage.parentThreadId !== null && thread.forkedFrom?.type === "node" - ? thread.lineage.parentThreadId - : undefined; + ); } /** @@ -109,39 +85,25 @@ function isTerminalRunUpdate(stored: OrchestrationV2StoredEvent): boolean { return ( stored.event.type === "run.updated" && FINISHED[stored.event.payload.status] !== undefined && - // Startup reconciliation settles runs left over from before the restart. + // Runtime reconciliation settles runs a restart cut short; the startup sweep reports them. !String(stored.commandId).startsWith("command:runtime-reconcile:") ); } /** - * Tells a thread when work its agent started in another thread ends or stops on - * the user. Threads launched with t3_thread_launch/create_threads and turns sent - * with t3_thread_send report back to the sending thread; a delegated child that - * waits on the user reports to its parent. Each notice is an ordinary queued - * notification message whose command and message ids derive from the run or - * request, so command receipts deliver it at most once. Only live events are - * observed, like the orchestrator's own terminal-run listener. + * Tells a thread when work its agent started in another thread ends. Threads + * launched with t3_thread_launch/create_threads and turns sent with + * t3_thread_send report back to the sending thread. Each notice is an ordinary + * queued notification message whose command and message ids derive from the + * run, so command receipts deliver it at most once. Live terminal runs are + * reported as they commit; after a restart, a sweep re-offers recently ended + * runs, which covers runs that ended unreported while the server was down or + * that restart reconciliation stopped. */ export const make = Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; - const eventStore = yield* EventStore.EventStoreV2; const orchestrator = yield* Orchestrator.OrchestratorV2; - const timers = yield* FiberMap.make(); - // Deadlines of armed wait timers, so `drain` can wait for the ones already due. - const timerDeadlines = yield* TxRef.make>(new Map()); - // Highest sequence each subscription has handled, so `drain` knows what it has not seen yet. - const observedSequence = { - "run.updated": yield* TxRef.make(0), - "runtime-request.updated": yield* TxRef.make(0), - }; - const setDeadline = (key: string, deadline: number | undefined) => - TxRef.update(timerDeadlines, (current) => { - const next = new Map(current); - if (deadline === undefined) next.delete(key); - else next.set(key, deadline); - return next; - }); + const sql = yield* SqlClient.SqlClient; const isReachable = (threadId: ThreadId) => orchestrator.getThreadRecords(threadId, []).pipe( @@ -149,214 +111,103 @@ export const make = Effect.gen(function* () { Effect.orElseSucceed(() => false), ); - const deliver = Effect.fn("ThreadReportBack.deliver")(function* (input: { - readonly key: string; - readonly source: OrchestrationV2AppThread; - readonly recipient: ThreadId; - readonly notification: OrchestrationV2Notification; - readonly text: string; - }) { - if (!(yield* isReachable(input.recipient))) return; - const id = `thread-report-back:${input.key}:${input.recipient}`; - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make(id), - messageId: MessageId.make(id), - threadId: input.recipient, - senderThreadId: input.source.id, - notification: input.notification, - text: input.text, - attachments: [], - // Notifications only queue: an idle recipient starts a turn now, a busy - // one hears about it after its current turn. - dispatchMode: { type: "queue_after_active" }, - createdBy: "agent", - creationSource: "server", - }); - }); - - const runSenders = (threadId: ThreadId, runId: RunId) => - orchestrator - .getThreadRecords(threadId, ["messages"], { - messageRunIds: [runId], - messageRoles: ["user"], - }) - .pipe(Effect.map(({ messages }) => agentSenders(threadId, messages))); - - const reportFinished = Effect.fn("ThreadReportBack.reportFinished")(function* ( - run: OrchestrationV2Run, - ) { + const reportFinished = Effect.fn("ThreadReportBack.reportFinished")(function* (run: FinishedRun) { const finished = FINISHED[run.status]; if (finished === undefined) return; - const { thread } = yield* orchestrator.getThreadRecords(run.threadId, []); - if (thread.deletedAt !== null || appOwnedSubagentParent(thread) !== undefined) return; + const { thread, messages } = yield* orchestrator.getThreadRecords(run.threadId, ["messages"], { + messageRunIds: [run.id], + messageRoles: ["user"], + }); + if (thread.deletedAt !== null || isAppOwnedSubagent(thread)) return; const title = titleOf(thread); - for (const recipient of yield* runSenders(thread.id, run.id)) { - yield* deliver({ - key: `finished:${thread.id}:${run.id}`, - source: thread, - recipient, + for (const recipient of agentSenders(thread.id, messages)) { + if (!(yield* isReachable(recipient))) continue; + const id = `thread-report-back:finished:${thread.id}:${run.id}:${recipient}`; + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(id), + messageId: MessageId.make(id), + threadId: recipient, + senderThreadId: thread.id, notification: { source: { kind: "subagent", childThreadId: thread.id }, outcome: finished.outcome, summary: `Thread "${title}" ${finished.verb}`, }, text: `${PREAMBLE} thread "${title}" (${thread.id}) ${finished.verb}. Read its result with t3_thread_read({threadId:"${thread.id}"}).`, + attachments: [], + // Notifications only queue: an idle recipient starts a turn now, a busy + // one hears about it after its current turn. + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", }); } }); - const reportWaiting = Effect.fn("ThreadReportBack.reportWaiting")(function* ( - work: Extract, - ) { - const records = yield* orchestrator.getThreadRecords( - work.threadId, - ["runtimeRequests", "runs"], - work.runId === undefined ? undefined : { runIds: [work.runId] }, - ); - const request = records.runtimeRequests.find((candidate) => candidate.id === work.requestId); - const waitingFor = request === undefined ? undefined : WAITING_FOR[request.kind]; - if (request?.status !== "pending" || waitingFor === undefined) return; - const thread = records.thread; - if (thread.deletedAt !== null) return; - const parent = appOwnedSubagentParent(thread); - const run = records.runs - .filter((candidate) => candidate.status !== "queued") - .toSorted((left, right) => left.ordinal - right.ordinal) - .at(-1); - const recipients = - parent !== undefined - ? [parent] - : run === undefined - ? [] - : yield* runSenders(thread.id, run.id); - const title = titleOf(thread); - for (const recipient of recipients) { - yield* deliver({ - key: `waiting:${thread.id}:${request.id}`, - source: thread, - recipient, - notification: { - source: { kind: "subagent", childThreadId: thread.id }, - outcome: "updated", - summary: `Thread "${title}" is waiting for the user`, - }, - text: `${PREAMBLE} thread "${title}" (${thread.id}) ${waitingFor}. Tell the user it needs them; do not answer or approve it on their behalf unless the user told you to. Inspect it with t3_thread_read({threadId:"${thread.id}"}).`, - }); - } - }); - - const logSkipped = - (message: string, fields: Record) => - (cause: Cause.Cause): Effect.Effect => - Cause.hasInterruptsOnly(cause) - ? Effect.failCause(cause) - : Effect.logWarning(message, { ...fields, cause }); - - const worker = yield* makeDrainableWorker((work: Work) => - (work.type === "finished" ? reportFinished(work.run) : reportWaiting(work)).pipe( - Effect.catchCause( - logSkipped("orchestration-v2.thread-report-back.failed", { - type: work.type, - threadId: work.type === "finished" ? work.run.threadId : work.threadId, - }), + const worker = yield* makeDrainableWorker((run: FinishedRun) => + reportFinished(run).pipe( + Effect.catchCauseIf( + (cause) => !Cause.hasInterruptsOnly(cause), + (cause) => + Effect.logWarning("orchestration-v2.thread-report-back.failed", { + threadId: run.threadId, + runId: run.id, + cause, + }), ), ), ); - const observe = (stored: OrchestrationV2StoredEvent) => - Effect.gen(function* () { - const event = stored.event; - if (event.type === "run.updated" && isTerminalRunUpdate(stored)) { - yield* worker.enqueue({ type: "finished", run: event.payload }); - } else if (event.type === "runtime-request.updated") { - const key = `${event.threadId}:${event.payload.id}`; - if (event.payload.status !== "pending") { - yield* FiberMap.remove(timers, key); - yield* setDeadline(key, undefined); - } else if ( - WAITING_FOR[event.payload.kind] !== undefined && - !(yield* FiberMap.has(timers, key)) - ) { - // The deadline is fixed when the request is seen, not when the timer fiber starts. - const deadline = (yield* Clock.currentTimeMillis) + WAITING_NOTICE_DELAY_MS; - const work: Work = { - type: "waiting", - threadId: event.threadId, - requestId: event.payload.id, - runId: event.runId, - }; - yield* setDeadline(key, deadline); - yield* FiberMap.run( - timers, - key, - Clock.currentTimeMillis.pipe( - Effect.flatMap((now) => Effect.sleep(Duration.millis(Math.max(0, deadline - now)))), - Effect.andThen(worker.enqueue(work)), - Effect.andThen(setDeadline(key, undefined)), - ), - ); - } - } - if (event.type === "run.updated" || event.type === "runtime-request.updated") { - yield* TxRef.set(observedSequence[event.type], stored.sequence); - } - }); - const start = Effect.fn("ThreadReportBack.start")(function* () { const afterSequence = yield* eventSink.latestSequence().pipe(Effect.orDie); - for (const eventType of OBSERVED_TYPES) { - yield* TxRef.set(observedSequence[eventType], afterSequence); - } - yield* Stream.mergeAll( - OBSERVED_TYPES.map((eventType) => eventSink.stream({ afterSequence, eventType })), - { concurrency: "unbounded" }, - ).pipe( - Stream.runForEach(observe), - Effect.catchCause(logSkipped("orchestration-v2.thread-report-back.stream-failed", {})), + yield* eventSink.stream({ afterSequence, eventType: "run.updated" }).pipe( + Stream.runForEach((stored) => + stored.event.type === "run.updated" && isTerminalRunUpdate(stored) + ? worker.enqueue(stored.event.payload) + : Effect.void, + ), + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.thread-report-back.stream-failed", { cause }), + ), Effect.forkScoped, ); }); - /** Waits until the subscription for `eventType` has handled its last committed event through `latest`. */ - const awaitObserved = (eventType: (typeof OBSERVED_TYPES)[number], latest: number) => - Effect.gen(function* () { - const observed = yield* TxRef.get(observedSequence[eventType]); - const unobserved = yield* eventStore - .read({ afterSequence: observed, throughSequence: latest, eventType }) - .pipe(Stream.runLast, Effect.orDie); - if (Option.isNone(unobserved)) return; - const target = unobserved.value.sequence; - yield* TxRef.get(observedSequence[eventType]).pipe( - Effect.tap((current) => (current < target ? Effect.txRetry : Effect.void)), - Effect.tx, - ); - }); - - /** - * Resolves once every committed event of the observed types was handled and - * its work, including wait timers already due, finished. Timers not yet due - * do not block. - */ - const drain = Effect.gen(function* () { - while (true) { - const latest = yield* eventSink.latestSequence().pipe(Effect.orDie); - for (const eventType of OBSERVED_TYPES) { - yield* awaitObserved(eventType, latest); - } - yield* Effect.gen(function* () { - const now = yield* Clock.currentTimeMillis; - const deadlines = yield* TxRef.get(timerDeadlines); - if ([...deadlines.values()].some((deadline) => deadline <= now)) { - return yield* Effect.txRetry; - } - }).pipe(Effect.tx); - yield* worker.drain; - if ((yield* eventSink.latestSequence().pipe(Effect.orDie)) === latest) return; - } - }); + /** Queues every recently ended run another thread's agent started; receipts drop repeats. */ + const sweep = Effect.gen(function* () { + const since = DateTime.formatIso( + DateTime.subtractDuration(yield* DateTime.now, SWEEP_LOOKBACK), + ); + const runs = yield* sql` + SELECT run.run_id AS id, run.thread_id AS "threadId", run.status + FROM orchestration_v2_projection_runs AS run + WHERE run.status IN ('completed', 'failed', 'cancelled', 'interrupted', 'rolled_back') + AND run.completed_at >= ${since} + AND EXISTS ( + SELECT 1 FROM orchestration_v2_projection_messages AS message + WHERE message.run_id = run.run_id + AND message.role = 'user' + AND json_extract(message.payload_json, '$.createdBy') = 'agent' + AND json_extract(message.payload_json, '$.senderThreadId') IS NOT NULL + ) + ORDER BY run.completed_at + `; + yield* Effect.forEach(runs, worker.enqueue, { discard: true }); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.thread-report-back.sweep-failed", { cause }), + ), + ); - return { start, drain }; + return { start, sweep, drain: worker.drain }; }); -export const layer = Layer.effectDiscard(make.pipe(Effect.flatMap((reactor) => reactor.start()))); +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const reactor = yield* make; + yield* reactor.start(); + // After startup recovery, so the runs it stopped are already settled. + yield* forkParked(reactor.sweep); + }), +); diff --git a/docs/operations/personal-fork.md b/docs/operations/personal-fork.md index c3428630af30d..74b3c9cac80e0 100644 --- a/docs/operations/personal-fork.md +++ b/docs/operations/personal-fork.md @@ -74,7 +74,7 @@ do not reimplement the project-folder behavior already in official v2. | Durable `/t3-goal` loops, fresh iterations, pause/resume/stop and handoff | [GoalLoopWorker](../../apps/server/src/orchestration-v2/GoalLoopWorker.ts), command/event contracts, projection and provider turn lifecycle | `GoalState.test.ts`, `GoalLoopWorker.test.ts`, `GoalLoop.test.ts`; start/steer/stop, usage-limit recovery, restart/replay and no goal inheritance into ordinary subagents | | Agent-drafted goal proposals that run only after the user starts them | [GoalState](../../apps/server/src/orchestration-v2/GoalState.ts), MCP goal toolkit, ChatView/composer and goal banner | Goal proposal coverage in `GoalState.test.ts` and `GoalLoop.test.ts`, plus `goalPresentation.test.ts`; proposing starts no turn, dismissal is repeatable, Start creates exactly one goal | | Goal replies/control channel, nested iteration rows, unfolding parked iterations and flat sidebar sorting | [Sidebar.logic](../../apps/web/src/components/Sidebar.logic.ts), client state and settings schemas | `Sidebar.logic.test.ts` and `packages/contracts/src/settings.test.ts`; persisted `last_activity`, goal/iteration visibility and selection, including stopped/parked goals | -| Launching ordinary threads from auto mode and receiving completion/waiting reports | [ThreadReportBack](../../apps/server/src/orchestration-v2/ThreadReportBack.ts), project/thread MCP handlers, launch authorization and notifications | `ThreadReportBack.test.ts`, project toolkit tests and `OrchestratorMcpToolkit.integration.test.ts`; correct workspace binding, bounded runtime permissions and one report to the initiating thread | +| Launching ordinary threads from auto mode and receiving completion reports, also after a restart | [ThreadReportBack](../../apps/server/src/orchestration-v2/ThreadReportBack.ts), project/thread MCP handlers, launch authorization and notifications | `ThreadReportBack.test.ts`, project toolkit tests and `OrchestratorMcpToolkit.integration.test.ts`; correct workspace binding, bounded runtime permissions and one report to the initiating thread | | Cursor questions and T3 tools in supported permission modes | [CursorT3Tools](../../apps/server/src/orchestration-v2/Adapters/CursorT3Tools.ts), Cursor adapter, MCP bridge and RPC authorization | `CursorT3Tools.test.ts` and `CursorAdapterV2.test.ts`; question/reply completion, restricted-mode tool availability and cancellation | | One fork release and coordinated updates, stable signing identity and T3 Connect configuration | [Fork release workflow](../../.github/workflows/fork-release.yml), `fork/update*.sh`, signing, server update checks and sidebar update UI | `ForkUpdate.test.ts`, update pill logic tests, built artifact inspection and installation/reconnect evidence on each Mac; same accepted commit/version, valid Connect configuration and signing identity | | Desktop shutdown finishes and stalled shutdowns leave useful diagnostics | [DesktopLifecycle](../../apps/desktop/src/app/DesktopLifecycle.ts), app/server teardown and updater handoff | `DesktopLifecycle.test.ts` and a real quit/update pass after active turns finish; preserve in-flight work and verify relaunch/reconnection separately | diff --git a/packages/provider-core/src/server/orchestrationInstructions.ts b/packages/provider-core/src/server/orchestrationInstructions.ts index 27f789dfc89fc..574ca4df559da 100644 --- a/packages/provider-core/src/server/orchestrationInstructions.ts +++ b/packages/provider-core/src/server/orchestrationInstructions.ts @@ -8,6 +8,7 @@ The \`t3-code\` MCP server provides app-owned orchestration. Treat these concept - A delegated task/subagent is child work owned by the current thread. Use \`orchestrator_capabilities\` to discover the current provider/model IDs from the same live catalog as the composer, including configured custom models. Do not treat a native tool's model list as the full list of available subagent models. Prefer native subagent tools for same-provider work only when they support the chosen model. Use \`delegate_task\` with that provider instance and model when native tools cannot, including for same-provider work. Also use \`delegate_task\` for cross-provider or explicitly T3-owned child tasks. Retain each returned \`taskId\`, and use \`task_status\` or \`task_cancel\` to manage it. The returned \`childThreadId\` is backing storage for the subagent, not the target for starting another delegated review round. - \`t3_thread_launch\` and \`create_threads\` create ordinary top-level T3 conversations. Use them only when the user explicitly asks for separate/new/top-level threads or conversations. Never use them merely because the user said "subagent" or requested parallel delegated work. +- Threads started or sent work with \`t3_thread_launch\`, \`create_threads\` or \`t3_thread_send\` report back to the sending thread when their turn ends, so do not block on \`t3_thread_wait\` for them. - For every T3 delegated review round, call \`delegate_task\` again. Include the original brief, prior findings, responses, and unresolved objections in each new task prompt. Track each round by its own \`taskId\`. Use a distinct \`clientRequestId\` per round, stable across retries of that round. Do not use \`t3_thread_send\` on \`childThreadId\` to continue a delegated review. - \`schedule_task\` creates persistent recurring work in the app scheduler. Pass \`schedule\` as a structured object, never as JSON text: \`{"type":"interval","everyMs":3600000}\` for an interval, or \`{"type":"fixed_time","timeOfDay":"09:00","weekdays":[1,2,3,4,5]}\` for a wall-clock schedule, or \`{"type":"webhook"}\` to run on each request to the returned \`webhookUrl\` (the run sees the request only through \`{{body.path}}\`-style placeholders in the prompt). By default runs return to the current thread, which suits orchestrating: each trigger arrives here and you delegate or dedupe; set \`bindToCurrentThread=false\` only when the user wants a fresh thread for every run. After scheduling a timer, report the returned cadence and next run time; for a webhook, report its \`webhookUrl\`, or say T3 Connect remote access is needed if it is missing. - \`t3_goal_propose\` drafts a goal: a long-running loop in which T3 Code keeps starting fresh agent iterations until a finish line is met. Use it when the user asks for a goal (including a message that starts with \`/t3-goal\`) and on your own when a task clearly needs more than one session, such as many dependent steps, long build or CI cycles, or iterating until tests pass. Infer the objective, finish line and background from the conversation rather than asking. The user starts the goal with one click, so after proposing, say so briefly and end your turn.