diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 180f053f111f4..c82b24c7f98b2 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -43,6 +43,7 @@ export const RPC_REQUIRED_SCOPES = { [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.getNativeChildWork]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.stopWorkflow]: AuthOrchestrationOperateScope, + [ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask]: AuthOrchestrationOperateScope, [ORCHESTRATION_V2_WS_METHODS.forceStopThread]: AuthOrchestrationOperateScope, [ORCHESTRATION_V2_WS_METHODS.getTurnDiff]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff]: AuthOrchestrationReadScope, @@ -61,6 +62,7 @@ export const RPC_REQUIRED_SCOPES = { [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.subscribeCommandOutput]: AuthOrchestrationReadScope, + [ORCHESTRATION_V2_WS_METHODS.subscribeBackgroundTaskOutput]: AuthOrchestrationReadScope, [ORCHESTRATION_V2_WS_METHODS.subscribePromptSuggestion]: AuthOrchestrationReadScope, [WS_METHODS.projectsMutate]: AuthOrchestrationOperateScope, [WS_METHODS.serverProbe]: AuthOrchestrationReadScope, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 9e29b0fdedc95..4fd3894bc7a34 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -259,6 +259,7 @@ export const make = Effect.gen(function* () { threadProjectFromScratch: true, threadTitleRegeneration: true, commandOutputStreaming: true, + backgroundTaskControl: true, threadVisitedTracking: true, threadPullRequests: true, threadPullRequestWatch: true, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index f9a861cca23fb..f928e97018cab 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -15,6 +15,7 @@ const RPC_AGGREGATES = { [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: "orchestration", [ORCHESTRATION_V2_WS_METHODS.getNativeChildWork]: "orchestration", [ORCHESTRATION_V2_WS_METHODS.stopWorkflow]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask]: "orchestration", [ORCHESTRATION_V2_WS_METHODS.forceStopThread]: "orchestration", [ORCHESTRATION_V2_WS_METHODS.getTurnItem]: "orchestration", [ORCHESTRATION_V2_WS_METHODS.getTurnDiff]: "orchestration", @@ -33,6 +34,7 @@ const RPC_AGGREGATES = { [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: "orchestrationV2", [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: "orchestrationV2", [ORCHESTRATION_V2_WS_METHODS.subscribeCommandOutput]: "orchestrationV2", + [ORCHESTRATION_V2_WS_METHODS.subscribeBackgroundTaskOutput]: "orchestrationV2", [ORCHESTRATION_V2_WS_METHODS.subscribePromptSuggestion]: "orchestrationV2", [WS_METHODS.projectsMutate]: "orchestration", [WS_METHODS.serverProbe]: "server", diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index a17d4d3d9f3ab..b7627b2ba4568 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -10826,7 +10826,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { const roster = providerThreadRosterEvents(harness.events).at(-1)?.providerThread .pendingBackgroundTasks; assert.deepEqual(roster, [ - { taskId: longRunning.taskId, kind: "monitor", description: "Monitor 0" }, + { + taskId: longRunning.taskId, + kind: "monitor", + description: "Monitor 0", + individualControl: true, + }, ]); }).pipe( Effect.provide( @@ -11424,6 +11429,144 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // Windows paths need no uid, so this runs on every host. + it.effect("follows and stops one background Bash task the roster lists", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const tmp = yield* fileSystem.makeTempDirectoryScoped({ + prefix: "t3-claude-background-output-", + }); + const stoppedTasks: Array = []; + const harness = yield* makeWakeHarnessWithOptions({ + environment: { CLAUDE_CODE_TMPDIR: tmp }, + stopTask: (taskId) => + Effect.sync(() => { + stoppedTasks.push(taskId); + }), + }); + const taskId = "dev-server-task"; + const tasks = path.join(tmp, "claude", "project", WAKE_NATIVE_SESSION, "tasks"); + yield* fileSystem.makeDirectory(tasks, { recursive: true }); + yield* fileSystem.writeFileString(path.join(tasks, `${taskId}.output`), "ready on :5173\n"); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-background-output"), + text: "Start the dev server in the background.", + attachments: [], + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: taskId, + description: "vp run dev", + is_backgrounded: true, + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000000b01", + session_id: WAKE_NATIVE_SESSION, + }), + ); + while ( + !(yield* Queue.take( + harness.providerThreadReceipts, + )).providerThread.pendingBackgroundTasks?.some((task) => task.taskId === taskId) + ); + + const output = harness.runtime.backgroundTaskOutput!; + const stop = harness.runtime.stopBackgroundTask!; + const target = { providerThread: harness.providerThread, taskId }; + assert.deepEqual( + Array.from(yield* Stream.runCollect(output(target).pipe(Stream.take(1)))), + ["ready on :5173\n"], + ); + + const missing = { ...target, taskId: "already-finished" }; + assert.isTrue(Exit.isFailure(yield* Stream.runCollect(output(missing)).pipe(Effect.exit))); + assert.isTrue(Exit.isFailure(yield* stop(missing).pipe(Effect.exit))); + yield* stop(target); + assert.deepEqual(stoppedTasks, [taskId]); + + // A command that sends its stdout to a file is followed there, not in Claude's capture. + const logFile = path.join(tmp, "dev.log"); + yield* fileSystem.writeFileString(logFile, "listening\n"); + for (const frame of [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: "msg_redirected_bash", + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: "toolu_redirected_bash", + name: "Bash", + input: { + command: `vp run dev > "${logFile}" 2>&1`, + run_in_background: true, + }, + }, + ], + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 1 }, + }, + parent_tool_use_id: null, + uuid: "00000000-0000-4000-8000-000000000b02", + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: "redirected-task", + tool_use_id: "toolu_redirected_bash", + description: "vp run dev", + is_backgrounded: true, + task_type: "local_bash", + uuid: "00000000-0000-4000-8000-000000000b03", + session_id: WAKE_NATIVE_SESSION, + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + let listed: OrchestrationV2PendingBackgroundTask | undefined; + while (listed === undefined) { + listed = (yield* Queue.take( + harness.providerThreadReceipts, + )).providerThread.pendingBackgroundTasks?.find( + (task) => task.taskId === "redirected-task", + ); + } + // Clients label the output with the file the server follows. + assert.equal(listed.outputFile, logFile); + const redirected = { ...target, taskId: "redirected-task" }; + assert.deepEqual( + Array.from(yield* Stream.runCollect(output(redirected).pipe(Stream.take(1)))), + ["listening\n"], + ); + }).pipe( + Effect.provideService(HostProcess.Platform, "win32"), + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + it.effect("lets the native SDK acknowledge Stop before closing its own query", () => Effect.scoped( Effect.gen(function* () { @@ -12110,7 +12253,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { (event) => event.providerThread.id === providerThreadA.id, )?.providerThread.pendingBackgroundTasks; assert.deepEqual(rosterAAfterSettle ?? [], [ - { taskId: taskA, description: "work on A", kind: "command" }, + { taskId: taskA, description: "work on A", kind: "command", individualControl: true }, ]); assert.isTrue(yield* hasPendingBackgroundWork); assert.isTrue(yield* hasPendingBackgroundWorkForThread(providerThreadA)); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index d677c5e68c70d..beb1e682e38c8 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -5,7 +5,11 @@ import { toolOutputImages, } from "@t3tools/shared/toolOutput"; import { makeProviderTextDeltaCoalescer } from "./ProviderTextDeltaCoalescer.ts"; -import { claudeTaskOutputRoot, tailClaudeTaskOutput } from "./ClaudeTaskOutputTail.ts"; +import { + claudeTaskOutputRoot, + tailClaudeTaskOutput, + tailOutputFile, +} from "./ClaudeTaskOutputTail.ts"; import { dynamicToolTitle, formatReadToolLabel, @@ -71,8 +75,10 @@ import type { } from "@anthropic-ai/claude-agent-sdk/sdk-tools"; // @effect-diagnostics-next-line nodeBuiltinImport:off -- the SDK takes a Node ChildProcess for the CLI. import * as NodeChildProcess from "node:child_process"; +import * as NodeOS from "node:os"; import * as NodeUtil from "node:util"; import { parseCliArgs } from "@t3tools/shared/cliArgs"; +import { shellStdoutFile, type ShellStdoutFile } from "@t3tools/shared/shellStdoutFile"; import { fromLenientJson } from "@t3tools/shared/schemaJson"; import { claudeMcpToolName, @@ -261,6 +267,8 @@ function claudePromptCacheTtlMs( // Below that, Claude compacts 33,000 tokens short of the window: Claude Code 2.1.305's // getContextUsage reports a 150,000 window as a 117,000 threshold, for every catalog model. const CLAUDE_AUTO_COMPACT_RESERVE_TOKENS = 33_000; +/** Bounds remembered Bash output locations; only tasks still running are ever read. */ +const MAX_TASK_OUTPUT_FILES = 256; function claudeAutoCompactThreshold(autoCompactWindow: string, maxTokens: number | null) { const windowTokens = Number(autoCompactWindow); @@ -3237,6 +3245,8 @@ function claudePendingBackgroundTask(input: { }): OrchestrationV2PendingBackgroundTask { return { taskId: input.taskId, + // Every roster task is a local_bash task, which stop_task ends and which writes an output file. + individualControl: true, kind: input.startedByMonitor ? "monitor" : ((input.taskType === null @@ -4526,6 +4536,8 @@ interface ClaudeLiveQueryContext { // Per process, not per session: a relative temp dir follows the cwd this CLI // was launched in, and a later turn's cwd can differ. readonly taskOutputRoot: string | undefined; + // Where Bash starts, for relative paths a command redirects its output to. + readonly bashCwd: string | undefined; } interface ActiveClaudeToolCall { @@ -5195,6 +5207,35 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( string, { readonly fiber: Fiber.Fiber; readonly stop: Deferred.Deferred } >(); + // Where each Bash task's output lands, so a viewer can follow one left running: the file + // its command redirects stdout to, else Claude's own capture. + const taskOutputFiles = new Map< + string, + | { readonly kind: "redirect"; readonly filePath: string } + | { readonly kind: "claude"; readonly root: string; readonly sessionId: string } + >(); + // Git Bash on Windows spells C:\ as /c/ and mounts /tmp at the Windows temp dir. Its other + // POSIX roots live inside its install, so those paths stay unresolved. + const windowsPath = (target: string) => { + if (hostPlatform !== "win32" || !target.startsWith("/")) return target; + const drive = /^\/([a-zA-Z])(\/.*)?$/.exec(target); + if (drive) return `${drive[1]}:${drive[2] ?? "/"}`; + const temp = /^\/tmp(\/.*)?$/.exec(target); + if (!temp) return undefined; + const tempDir = + adapterOptions.environment.TEMP || adapterOptions.environment.TMP || NodeOS.tmpdir(); + return path.join(tempDir, temp[1] ?? ""); + }; + const resolveStdoutFile = (stdout: ShellStdoutFile, cwd: string | undefined) => { + let directory = cwd ?? "."; + for (const change of stdout.cwdChanges) { + const next = windowsPath(change); + if (next === undefined) return undefined; + directory = path.resolve(directory, next); + } + const file = windowsPath(stdout.path); + return file === undefined ? undefined : path.resolve(directory, file); + }; // Waits for the tail's last read so its output lands before the call settles. const stopTaskOutputTail = (toolUseId: string) => Effect.suspend(() => { @@ -8925,7 +8966,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( launch?.classification.itemType === "command_execution" ? firstStringInputField(launch.input, ["command"]) : undefined; + const output = taskOutputFiles.get(message.task_id); yield* upsertPendingBackgroundTask(input.nativeThreadId, { + ...(output?.kind === "redirect" ? { outputFile: output.filePath } : {}), ...(launch === undefined || command === undefined ? {} : { @@ -9714,6 +9757,37 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( if (message.type === "system" && message.subtype === "task_started") { if (isClaudeNonSubagentTask(message)) { + if (message.task_type === "local_bash") { + const launch = + message.tool_use_id === undefined + ? undefined + : findToolCall(context, message.tool_use_id); + const command = + launch?.classification.itemType === "command_execution" + ? firstStringInputField(launch.input, ["command"]) + : undefined; + const stdout = command === undefined ? undefined : shellStdoutFile(command); + // A file it cannot place falls back to Claude's capture rather than a guessed path. + const redirect = + stdout === undefined ? undefined : resolveStdoutFile(stdout, liveQuery.bashCwd); + const output = + redirect !== undefined + ? { kind: "redirect" as const, filePath: redirect } + : liveQuery.taskOutputRoot !== undefined + ? { + kind: "claude" as const, + root: liveQuery.taskOutputRoot, + sessionId: message.session_id, + } + : undefined; + if (output !== undefined) { + if (taskOutputFiles.size >= MAX_TASK_OUTPUT_FILES) { + const [oldest] = taskOutputFiles.keys(); + if (oldest !== undefined) taskOutputFiles.delete(oldest); + } + taskOutputFiles.set(message.task_id, output); + } + } const tailedCall = message.tool_use_id === undefined ? undefined @@ -11321,6 +11395,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( resolvedTurnCwd, hostPlatform, ), + bashCwd: resolvedTurnCwd ?? undefined, }; // One step, so the replaced process cannot exit between the install and the handoff. yield* Ref.update(queryContext, (replaced) => { @@ -12359,6 +12434,77 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ), ); }), + stopBackgroundTask: Effect.fn("ClaudeAdapterV2.stopBackgroundTask")(function* ({ + providerThread, + taskId, + }) { + const liveQuery = yield* Ref.get(queryContext); + const nativeThreadId = providerThread.nativeThreadRef?.nativeId; + if ( + liveQuery === null || + liveQuery.stopping || + nativeThreadId == null || + liveQuery.nativeThreadId !== nativeThreadId || + !(yield* hasPendingBackgroundTaskOnNativeThread(nativeThreadId, taskId)) + ) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CLAUDE_PROVIDER, + detail: "No running background task belongs to this provider thread", + }); + } + // Claude reports the stop through task_notification, which updates the roster. + yield* liveQuery.query.stopTask(taskId).pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CLAUDE_PROVIDER, + detail: "Background task stop request failed", + cause, + }), + ), + ); + }), + backgroundTaskOutput: ({ providerThread, taskId }) => + Stream.callback( + (queue) => + Effect.gen(function* () { + const nativeThreadId = providerThread.nativeThreadRef?.nativeId; + const file = taskOutputFiles.get(taskId); + if ( + nativeThreadId == null || + file === undefined || + !(yield* hasPendingBackgroundTaskOnNativeThread(nativeThreadId, taskId)) + ) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CLAUDE_PROVIDER, + detail: "No readable background task output belongs to this provider thread", + }); + } + const stop = yield* Deferred.make(); + yield* Effect.addFinalizer(() => Deferred.succeed(stop, undefined)); + const onChunk = (chunk: string) => Queue.offer(queue, chunk).pipe(Effect.asVoid); + // Only a path from a command this thread's agent ran is ever read here. + // Claude's capture ends the stream when Claude never writes it. + yield* ( + file.kind === "redirect" + ? tailOutputFile({ filePath: file.filePath, stop, onChunk }) + : tailClaudeTaskOutput({ + root: file.root, + sessionId: file.sessionId, + taskId, + stop, + onChunk, + }) + ).pipe( + Effect.andThen(Queue.end(queue)), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.forkScoped, + ); + }), + // A slow viewer pauses the tail instead of queueing reads without bound. + { bufferSize: 16, strategy: "suspend" }, + ), awaitBackgroundTasks: Effect.fn("ClaudeAdapterV2.awaitBackgroundTasks")( function* ({ providerThread, taskIds, awaited }) { const nativeThreadId = providerThread.nativeThreadRef?.nativeId; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.test.ts index b70b7322d5e63..2431bdfa1942b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.test.ts @@ -22,6 +22,39 @@ const makeTaskFile = Effect.gen(function* () { // TestClock cannot interleave with real filesystem I/O; tests that wait on a poll use the live clock. describe("tailClaudeTaskOutput", () => { + it.effect("reads nothing for ids that would leave the session's tasks folder", () => + Effect.scoped( + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const { root } = yield* makeTaskFile; + // Where `../../secret` as a task id, and `..` as a session id, would lead. + for (const outside of [ + path.join(root, "-private-tmp-project", "secret.output"), + path.join(root, "tasks", "secret.output"), + ]) { + yield* fileSystem.makeDirectory(path.dirname(outside), { recursive: true }); + yield* fileSystem.writeFileString(outside, "secret\n"); + } + const stop = yield* Deferred.make(); + yield* Deferred.succeed(stop, undefined); + const chunks: Array = []; + for (const ids of [ + { sessionId: SESSION_ID, taskId: "../../secret" }, + { sessionId: "..", taskId: "secret" }, + ]) { + yield* tailClaudeTaskOutput({ + root, + ...ids, + stop, + onChunk: (chunk) => Effect.sync(() => chunks.push(chunk)), + }); + } + assert.deepEqual(chunks, []); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + it.live("follows a file that appears after the task starts, across split characters", () => Effect.scoped( Effect.gen(function* () { @@ -164,4 +197,19 @@ describe("claudeTaskOutputRoot", () => { ); }).pipe(Effect.provide(NodeServices.layer)), ); + + it.effect("uses Claude Code's shared claude folder in the Windows temp dir", () => + Effect.gen(function* () { + const path = yield* Path.Path; + const temp = path.resolve("Users", "me", "AppData", "Local", "Temp"); + const root = (environment: Record) => + claudeTaskOutputRoot(environment, path, null, "win32"); + assert.equal(root({ TEMP: temp, TMP: "ignored" }), path.join(temp, "claude")); + assert.equal(root({ TMP: temp }), path.join(temp, "claude")); + assert.equal( + root({ CLAUDE_CODE_TMPDIR: path.resolve("srv"), TEMP: temp }), + path.resolve("srv", "claude"), + ); + }).pipe(Effect.provide(NodeServices.layer)), + ); }); diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.ts index 77505d24a5bbf..16a60fa4453bd 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeTaskOutputTail.ts @@ -1,3 +1,5 @@ +import * as NodeOS from "node:os"; + import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; @@ -10,16 +12,26 @@ import * as Path from "effect/Path"; const CLAUDE_TASK_OUTPUT_POLL_MS = 250; /** Bound discovery when Claude never writes the file. */ const MAX_LOCATE_ATTEMPTS = 40; +const CLAUDE_TASK_PATH_ID = /^[\w-]+$/; /** Skip ahead of fast writers; only the tail is displayed. */ const MAX_READ_BYTES = 256 * 1024; -/** Claude Code's per-user temp root, or none where it has no uid (Windows). */ +/** Claude Code's temp root: per user (`claude-`) on POSIX, and `claude` on Windows, which has no uid. */ export function claudeTaskOutputRoot( environment: Readonly> | undefined, path: Path.Path, cwd: string | null, platform: NodeJS.Platform, ): string | undefined { + if (platform === "win32") { + // Claude Code's `os.tmpdir()`, read from the environment it was launched with. + const base = + environment?.CLAUDE_CODE_TMPDIR?.trim() || + environment?.TEMP || + environment?.TMP || + NodeOS.tmpdir(); + return path.resolve(cwd ?? ".", base, "claude"); + } const uid = process.getuid?.(); if (uid === undefined) return undefined; // Claude Code's macOS build falls back to /tmp and ignores the TMPDIR that @@ -39,6 +51,10 @@ const locate = Effect.fn("ClaudeTaskOutputTail.locate")(function* (input: { readonly sessionId: string; readonly taskId: string; }) { + // Both ids come from the provider stream; anything but a plain id could leave the root. + if (!CLAUDE_TASK_PATH_ID.test(input.sessionId) || !CLAUDE_TASK_PATH_ID.test(input.taskId)) { + return Option.none(); + } const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; const projects = yield* fileSystem.readDirectory(input.root).pipe(Effect.orElseSucceed(() => [])); @@ -65,21 +81,34 @@ export const tailClaudeTaskOutput = Effect.fn("ClaudeTaskOutputTail.tail")(funct readonly stop: Deferred.Deferred; readonly onChunk: (chunk: string) => Effect.Effect; }) { - const fileSystem = yield* FileSystem.FileSystem; - const pause = Deferred.await(input.stop).pipe( - Effect.as(true), - Effect.timeoutOrElse({ - duration: CLAUDE_TASK_OUTPUT_POLL_MS, - orElse: () => Effect.succeed(false), - }), - ); + const pause = pauseUntilStop(input.stop); let file = Option.none(); for (let attempt = 0; Option.isNone(file); attempt += 1) { file = yield* locate(input); if (Option.isSome(file)) break; if (attempt + 1 >= MAX_LOCATE_ATTEMPTS || (yield* pause)) return; } - const filePath = file.value; + yield* tailOutputFile({ filePath: file.value, stop: input.stop, onChunk: input.onChunk }); +}); + +const pauseUntilStop = (stop: Deferred.Deferred) => + Deferred.await(stop).pipe( + Effect.as(true), + Effect.timeoutOrElse({ + duration: CLAUDE_TASK_OUTPUT_POLL_MS, + orElse: () => Effect.succeed(false), + }), + ); + +/** Follows a file from its recent tail until `stop`, waiting for it to appear and restarting when it is truncated. */ +export const tailOutputFile = Effect.fn("ClaudeTaskOutputTail.tailFile")(function* (input: { + readonly filePath: string; + readonly stop: Deferred.Deferred; + readonly onChunk: (chunk: string) => Effect.Effect; +}) { + const fileSystem = yield* FileSystem.FileSystem; + const pause = pauseUntilStop(input.stop); + const filePath = input.filePath; let decoder = new TextDecoder("utf-8", { fatal: false }); let offset = 0; let last = yield* Deferred.isDone(input.stop); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index 742616f27bd3e..09ed550f3f8a6 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -558,6 +558,130 @@ it.effect.each(["live", "no-session", "unsupported-runtime"] as const)( }), ); +// One provider thread listing a running dev server, served by a runtime with the given methods. +const backgroundTaskService = Effect.fn("test.backgroundTaskService")(function* ( + runtimeMethods: Partial | null, +) { + const threadId = ThreadId.make("thread:background-task"); + const providerThreadId = ProviderThreadId.make("provider-thread:background-task"); + const providerSessionId = ProviderSessionId.make("provider-session:background-task"); + const driver = ProviderDriverKind.make("claudeAgent"); + const now = yield* DateTime.now; + const providerThread: OrchestrationV2ProviderThread = { + id: providerThreadId, + driver, + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + pendingBackgroundTasks: [{ taskId: "dev-server", kind: "command", individualControl: true }], + contextUsage: null, + nativeMetadata: null, + createdAt: now, + updatedAt: now, + }; + const runtime = { + driver, + providerSessionId, + ...runtimeMethods, + } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadRecords: () => + Effect.succeed({ + thread: { id: threadId, deletedAt: null, activeProviderThreadId: providerThreadId }, + providerThreads: [providerThread], + } as unknown as OrchestrationV2ThreadProjection), + }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ + get: () => Effect.succeed(runtimeMethods === null ? Option.none() : Option.some(runtime)), + }), + ), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(Layer.merge(layerTest, ThreadCommandExecutor.layer)), + ); + return { service, threadId, providerThreadId }; +}); + +it.effect.each(["live", "not-running", "no-session", "unsupported-runtime"] as const)( + "stops one background task through the provider thread that lists it, with state %s", + (state) => + Effect.gen(function* () { + const stopped: Array = []; + const { service, threadId, providerThreadId } = yield* backgroundTaskService( + state === "no-session" + ? null + : state === "unsupported-runtime" + ? {} + : { + stopBackgroundTask: (input) => + Effect.sync(() => { + stopped.push(`${input.providerThread.id}:${input.taskId}`); + }), + }, + ); + const result = yield* service + .stopBackgroundTask({ + threadId, + taskId: state === "not-running" ? "finished" : "dev-server", + }) + .pipe(Effect.result); + + if (state === "live") { + expect(result).toMatchObject({ _tag: "Success" }); + expect(stopped).toEqual([`${providerThreadId}:dev-server`]); + } else { + expect(result).toMatchObject({ + failure: { + _tag: "OrchestrationV2StopBackgroundTaskError", + reason: { + "not-running": "not-running", + "no-session": "unavailable", + "unsupported-runtime": "unsupported", + }[state], + }, + }); + expect(stopped).toEqual([]); + } + }), +); + +it.effect("streams a background task's output in bounded frames and says when it ends", () => + Effect.gen(function* () { + const { service, threadId } = yield* backgroundTaskService({ + backgroundTaskOutput: () => Stream.make("ready\n", "x".repeat(20_000)), + }); + const frames = Array.from( + yield* Stream.runCollect( + service.subscribeBackgroundTaskOutput({ threadId, taskId: "dev-server" }), + ), + ); + + expect(frames.map(({ kind, running }) => [kind, running])).toEqual([ + ["replace", true], + ["append", true], + // A read larger than one append resyncs viewers from the bounded tail. + ["replace", true], + ["append", false], + ]); + expect(frames[1]?.text).toBe("ready\n"); + expect(frames[2]?.text).toBe(`ready\n${"x".repeat(20_000)}`); + }), +); + it.effect.each([ { status: "completed" as const, followUp: "none", settles: true }, { status: "failed" as const, followUp: "none", settles: false }, diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 3b1fe7b6a5e96..bd64e5964214a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -1,3 +1,8 @@ +import { + appendTerminalOutput, + EMPTY_TERMINAL_OUTPUT, + terminalOutputResumeText, +} from "@t3tools/shared/terminalOutput"; import type { OrchestrationV2SearchThreadInput, OrchestrationV2SearchThreadResult, @@ -15,6 +20,11 @@ import { MessageId, OrchestrationV2StopWorkflowError, type OrchestrationV2StopWorkflowInput, + OrchestrationV2StopBackgroundTaskError, + type OrchestrationV2StopBackgroundTaskInput, + OrchestrationV2CommandOutputError, + type OrchestrationV2CommandOutputFrame, + type OrchestrationV2SubscribeBackgroundTaskOutputInput, OrchestrationV2ForceStopThreadError, type OrchestrationV2ForceStopThreadInput, type OrchestrationV2ForceStopThreadResult, @@ -475,6 +485,12 @@ export interface ThreadManagementServiceShape { readonly forceStopThread: ( input: OrchestrationV2ForceStopThreadInput, ) => Effect.Effect; + readonly stopBackgroundTask: ( + input: OrchestrationV2StopBackgroundTaskInput, + ) => Effect.Effect; + readonly subscribeBackgroundTaskOutput: ( + input: OrchestrationV2SubscribeBackgroundTaskOutputInput, + ) => Stream.Stream; readonly awaitBackgroundTasks: (input: { readonly threadId: ThreadId; readonly taskIds: ReadonlyArray; @@ -574,6 +590,8 @@ function latestSteerableRun( const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000; const FORCE_STOP_REASON = "Force stopped by user"; const FORCE_STOP_RELEASE_WAIT = "10 seconds"; +/** Matches CommandOutputHub: larger reads go to clients as a bounded replace. */ +const BACKGROUND_TASK_OUTPUT_MAX_APPEND_CHARS = 16 * 1024; const isForceStopThreadError = Schema.is(OrchestrationV2ForceStopThreadError); const make = Effect.gen(function* () { @@ -1481,6 +1499,106 @@ const make = Effect.gen(function* () { return { matched, unmatched: input.taskIds.filter((taskId) => !matchedIds.has(taskId)) }; }); + // The provider thread whose roster lists the task, with its live runtime if any. + const findBackgroundTask = Effect.fn("ThreadManagementService.findBackgroundTask")( + function* (input: { readonly threadId: ThreadId; readonly taskId: string }) { + const records = yield* orchestrator.getThreadRecords(input.threadId, ["providerThreads"]); + const providerThread = records.providerThreads.find( + (candidate) => + candidate.appThreadId === input.threadId && + candidate.pendingBackgroundTasks?.some((task) => task.taskId === input.taskId), + ); + if (records.thread.deletedAt !== null || providerThread === undefined) return null; + const runtime = + providerThread.providerSessionId == null + ? Option.none() + : yield* sessions.get(providerThread.providerSessionId); + return { providerThread, runtime }; + }, + ); + + const stopBackgroundTask: ThreadManagementServiceShape["stopBackgroundTask"] = Effect.fn( + "ThreadManagementService.stopBackgroundTask", + )( + function* (input) { + const fail = (reason: OrchestrationV2StopBackgroundTaskError["reason"], cause?: unknown) => + new OrchestrationV2StopBackgroundTaskError({ + ...input, + reason, + ...(cause === undefined ? {} : { cause }), + }); + const found = yield* findBackgroundTask(input).pipe( + Effect.mapError((cause) => fail("unavailable", cause)), + ); + if (found === null) return yield* fail("not-running"); + if (Option.isNone(found.runtime)) return yield* fail("unavailable"); + const stop = found.runtime.value.stopBackgroundTask; + if (stop === undefined) return yield* fail("unsupported"); + // Bounded so a provider that never answers cannot hold the thread lock. + yield* stop({ providerThread: found.providerThread, taskId: input.taskId }).pipe( + Effect.timeout("15 seconds"), + Effect.mapError((cause) => fail("stop-failed", cause)), + ); + }, + (effect, input) => effect.pipe((effect) => threadDispatch.withLock(input.threadId, effect)), + ); + + const subscribeBackgroundTaskOutput: ThreadManagementServiceShape["subscribeBackgroundTaskOutput"] = + (input) => + Stream.unwrap( + Effect.gen(function* () { + const failure = (message: string, cause?: unknown) => + new OrchestrationV2CommandOutputError({ + threadId: input.threadId, + message, + ...(cause === undefined ? {} : { cause }), + }); + const found = yield* findBackgroundTask(input).pipe( + Effect.mapError((cause) => failure("Couldn't find the background task.", cause)), + ); + if (found === null) return yield* failure("This background task is no longer running."); + const read = Option.isSome(found.runtime) + ? found.runtime.value.backgroundTaskOutput + : undefined; + if (read === undefined) { + return yield* failure("This provider doesn't share background task output."); + } + let tail = EMPTY_TERMINAL_OUTPUT; + const frames = read({ providerThread: found.providerThread, taskId: input.taskId }).pipe( + Stream.map((chunk): OrchestrationV2CommandOutputFrame => { + tail = appendTerminalOutput(tail, chunk); + // A large first read (the file's recent tail) goes out bounded. + return chunk.length > BACKGROUND_TASK_OUTPUT_MAX_APPEND_CHARS + ? { + kind: "replace", + text: terminalOutputResumeText(tail), + truncated: tail.truncated, + running: true, + } + : { kind: "append", text: chunk, truncated: tail.truncated, running: true }; + }), + Stream.mapError((cause) => failure("Lost the background task's output.", cause)), + ); + // An opening empty frame lets a quiet command show as running rather than loading. + const opening: OrchestrationV2CommandOutputFrame = { + kind: "replace", + text: "", + truncated: false, + running: true, + }; + // A stream that ends (the provider never wrote output) says so instead of looking live. + const closing = Stream.suspend(() => + Stream.succeed({ + kind: "append", + text: "", + truncated: tail.truncated, + running: false, + }), + ); + return Stream.succeed(opening).pipe(Stream.concat(frames), Stream.concat(closing)); + }), + ); + const stopDelegatedTasks: ThreadManagementServiceShape["stopDelegatedTasks"] = Effect.fn( "ThreadManagementService.stopDelegatedTasks", )(function* (input) { @@ -1641,6 +1759,8 @@ const make = Effect.gen(function* () { stopDelegatedTasks, stopWorkflow, forceStopThread, + stopBackgroundTask, + subscribeBackgroundTaskOutput, awaitBackgroundTasks, getThreadEventSequence: orchestrator.getThreadEventSequence, recoverDelegatedTask: orchestrator.recoverDelegatedTask, diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_monitor_wake/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_monitor_wake/output.ts index c1c3d9a171b0f..16dc6de1e5ce9 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_monitor_wake/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_monitor_wake/output.ts @@ -40,6 +40,7 @@ export function assertClaudeBackgroundMonitorWakeOutput( taskId: MONITOR_TASK_ID, description: "Background monitor test", kind: "monitor", + individualControl: true, }); } assert.deepEqual(projection.providerThreads[0]?.pendingBackgroundTasks ?? [], []); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_wake/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_wake/output.ts index 416718035fcbb..afbffc5ff9f94 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_wake/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/claude_background_task_wake/output.ts @@ -80,6 +80,7 @@ export function assertClaudeBackgroundTaskWakeOutput( taskId: BACKGROUND_TASK_ID, description: "Background sleep test", kind: "command", + individualControl: true, }); assert.deepEqual(backgroundNotifications(projection), [ { diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 85b9b44f4daf7..862e379020fbf 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -256,6 +256,8 @@ const makeTestRelay = Effect.fnUntraced(function* ( settleThread: unused, interruptThread: unused, stopWorkflow: () => Effect.die("unused stopWorkflow"), + stopBackgroundTask: () => Effect.die("unused stopBackgroundTask"), + subscribeBackgroundTaskOutput: () => Stream.die("unused subscribeBackgroundTaskOutput"), forceStopThread: () => Effect.die("unused forceStopThread"), awaitBackgroundTasks: unused, stopDelegatedTasks: unused, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index f80a177e3191e..ed9a6d06e0d29 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1965,6 +1965,8 @@ export const layerWsRpc = ( ), ), [ORCHESTRATION_V2_WS_METHODS.stopWorkflow]: (input) => threadManagement.stopWorkflow(input), + [ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask]: (input) => + threadManagement.stopBackgroundTask(input), [ORCHESTRATION_V2_WS_METHODS.forceStopThread]: (input) => threadManagement.forceStopThread(input), [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: (input) => @@ -2211,6 +2213,8 @@ export const layerWsRpc = ( Effect.as(commandOutputHub.subscribe(input)), ), ), + [ORCHESTRATION_V2_WS_METHODS.subscribeBackgroundTaskOutput]: (input) => + threadManagement.subscribeBackgroundTaskOutput(input), [ORCHESTRATION_V2_WS_METHODS.subscribePromptSuggestion]: (input) => providerSessionManager.subscribePromptSuggestion(input.threadId), [WS_METHODS.scheduledTasksSubscribe]: (_input) => diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index c1f68878e9644..11b9cb8d45574 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -7,6 +7,7 @@ import { snapshotComposerSend, withMessageContext, deriveAvailableRevertTurnCounts, + deriveThreadRunningBackgroundTasks, } from "./ChatView.logic"; import { recallCheckoutIsRepo, rememberCheckoutIsRepo } from "./ChatView.logic"; import { ProviderDriverKind, type ServerProvider } from "@t3tools/contracts"; @@ -1239,3 +1240,29 @@ describe("Open in for the host environment", () => { } }); }); + +describe("deriveThreadRunningBackgroundTasks", () => { + it("lists the active provider thread's commands while a turn runs, without subagents", () => { + const devServer = { + taskId: "dev-server", + kind: "command" as const, + individualControl: true as const, + }; + const projection = { + thread: { activeProviderThreadId: "provider-thread:active" }, + runs: [{ id: "run:2", status: "running" }], + providerThreads: [ + { + id: "provider-thread:active", + pendingBackgroundTasks: [devServer, { taskId: "explorer", kind: "subagent" }], + }, + { + id: "provider-thread:earlier", + pendingBackgroundTasks: [{ taskId: "old-watcher", kind: "monitor" }], + }, + ], + } as unknown as Parameters[0]; + + expect(deriveThreadRunningBackgroundTasks(projection)).toEqual([devServer]); + }); +}); diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index d35b6ce28c433..b3f72f567526a 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -1,4 +1,8 @@ -import { type OrchestrationV2ProviderTurn, type RunAttemptId } from "@t3tools/contracts"; +import { + type OrchestrationV2PendingBackgroundTask, + type OrchestrationV2ProviderTurn, + type RunAttemptId, +} from "@t3tools/contracts"; import { deriveLatestThreadRun, deriveThreadRuntime, @@ -1465,6 +1469,25 @@ export function deriveThreadPendingBackgroundTasks(projection: ThreadStructure | ]; } +const NO_RUNNING_BACKGROUND_TASKS: ReadonlyArray = []; + +/** + * The agent's background commands as its provider reports them, without the hold rules of + * `deriveThreadPendingBackgroundTasks`: a dev server keeps running while a new turn works. + */ +export function deriveThreadRunningBackgroundTasks( + projection: ThreadStructure | null, +): ReadonlyArray { + if (projection === null) return NO_RUNNING_BACKGROUND_TASKS; + const activeId = projection.thread.activeProviderThreadId; + const tasks = projection.providerThreads + .filter((thread) => activeId == null || thread.id === activeId) + .flatMap((thread) => thread.pendingBackgroundTasks ?? []) + // Subagents are listed under Lineage. + .filter((task) => task.kind !== "subagent"); + return tasks.length === 0 ? NO_RUNNING_BACKGROUND_TASKS : tasks; +} + /** * Run ordinals the conversation can be reverted to: those whose provider turn belongs to the * active provider thread, plus 0 for the thread start. Empty without a live provider session diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index a14d24c947b35..bf2c25721f2c9 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -10,6 +10,7 @@ import { announcedRun, isDetailBehindShell } from "./chat/threadActivityAnnounce import { deriveAvailableRevertTurnCounts, deriveThreadPendingBackgroundTasks, + deriveThreadRunningBackgroundTasks, presentThreadShellFromProjection, } from "./ChatView.logic"; import { usePullRequestDraftThread } from "./chat/usePullRequestDraftThread"; @@ -1448,6 +1449,7 @@ export default function ChatView(props: ChatViewProps) { activeProviderThread, ); const pendingBackgroundTasks = deriveThreadPendingBackgroundTasks(serverProjection); + const runningBackgroundTasks = deriveThreadRunningBackgroundTasks(serverProjection); const activeWorkStartedAt = deriveActiveWorkStartedAt(activeActivityRun, activeRuntime, localDispatchStartedAt) ?? runlessWorkStartedAt; @@ -2397,6 +2399,7 @@ export default function ChatView(props: ChatViewProps) { showBranchMismatchBanner, threadStatusLine, shortcuts: threadLifecycleShortcuts, + backgroundWork, } = useThreadLifecycle({ environmentId, routeThreadKey, @@ -2416,6 +2419,7 @@ export default function ChatView(props: ChatViewProps) { runtimeMode, isWorking, pendingBackgroundTasks, + runningBackgroundTasks, turnItems: serverProjection?.turnItems ?? EMPTY_TURN_ITEMS, activePendingApproval, activePendingUserInput, @@ -3892,6 +3896,7 @@ export default function ChatView(props: ChatViewProps) { onAddProjectScript={saveProjectScriptStable} onUpdateProjectScript={updateProjectScriptStable} onDeleteProjectScript={deleteProjectScriptStable} + backgroundWork={backgroundWork} usage={{ providerInstanceId: activeProviderInstanceId, providers: providerStatuses, diff --git a/apps/web/src/components/chat/BackgroundWorkDetails.browser.test.tsx b/apps/web/src/components/chat/BackgroundWorkDetails.browser.test.tsx index 01311a60ecf4f..f249d03199eed 100644 --- a/apps/web/src/components/chat/BackgroundWorkDetails.browser.test.tsx +++ b/apps/web/src/components/chat/BackgroundWorkDetails.browser.test.tsx @@ -6,6 +6,7 @@ import { page } from "vite-plus/test/browser"; vi.mock(import("./CommandOutputPanel"), () => ({ useCommandOutputStreaming: () => true, CommandOutputPanel: () =>
Command output stream
, + BackgroundTaskOutputPanel: () =>
Background task output
, })); import { render } from "../../../test/render"; diff --git a/apps/web/src/components/chat/BackgroundWorkDetails.tsx b/apps/web/src/components/chat/BackgroundWorkDetails.tsx index 86df01be78f59..336ff884b6bd8 100644 --- a/apps/web/src/components/chat/BackgroundWorkDetails.tsx +++ b/apps/web/src/components/chat/BackgroundWorkDetails.tsx @@ -4,7 +4,7 @@ import type { ThreadId, } from "@t3tools/contracts"; import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; -import { useState } from "react"; +import { useState, type ReactElement, type ReactNode } from "react"; import { InlineButton } from "../ui/button"; import { @@ -15,8 +15,27 @@ import { DialogTitle, DialogTrigger, } from "../ui/dialog"; -import { CommandOutputPanel, useCommandOutputStreaming } from "./CommandOutputPanel"; +import { + BackgroundTaskOutputPanel, + CommandOutputPanel, + useCommandOutputStreaming, +} from "./CommandOutputPanel"; import { ShellCommandBlock } from "./ShellCommandBlock"; +import { useBackgroundTaskControl, useBackgroundTaskStop } from "./useBackgroundTaskStop"; + +/** The turn item that launched a background task, when the timeline still holds it. */ +export function backgroundTaskLaunch( + task: OrchestrationV2PendingBackgroundTask, + turnItems: ThreadStructure["turnItems"], +) { + const item = turnItems.findLast((candidate) => + task.itemId === undefined + ? (candidate.nativeItemRef?.nativeId || candidate.id) === task.taskId + : candidate.id === task.itemId, + ); + const command = task.command ?? (item?.type === "command_execution" ? item.input : undefined); + return { item, command }; +} export function BackgroundWorkDetails(props: { readonly environmentId: EnvironmentId; @@ -24,12 +43,18 @@ export function BackgroundWorkDetails(props: { readonly tasks: ReadonlyArray; readonly turnItems: ThreadStructure["turnItems"]; readonly onOpenThread: (threadId: ThreadId) => void; + readonly trigger?: ReactElement; + readonly children?: ReactNode; }) { const [open, setOpen] = useState(false); const streamsOutput = useCommandOutputStreaming(props.environmentId); + const followsTasks = useBackgroundTaskControl(props.environmentId); + const taskStop = useBackgroundTaskStop(props.environmentId, props.threadId); return ( - }>Details + }> + {props.children ?? "Details"} + Background work @@ -37,13 +62,7 @@ export function BackgroundWorkDetails(props: {
{props.tasks.map((task) => { - const item = props.turnItems.findLast((candidate) => - task.itemId === undefined - ? (candidate.nativeItemRef?.nativeId || candidate.id) === task.taskId - : candidate.id === task.itemId, - ); - const command = - task.command ?? (item?.type === "command_execution" ? item.input : undefined); + const { item, command } = backgroundTaskLaunch(task, props.turnItems); const description = task.description?.trim(); const title = description && description !== command?.trim() @@ -54,12 +73,24 @@ export function BackgroundWorkDetails(props: { const itemId = task.itemId ?? (item?.type === "command_execution" ? item.id : undefined); const childThreadId = task.kind === "subagent" ? task.childThreadId : undefined; + const isCommand = task.kind !== "subagent"; + const controlled = isCommand && task.individualControl === true; + const followedFile = task.outputFile; + const stopping = taskStop.stopping.has(task.taskId); return (

{title}

- + {task.awaited ? "Waiting" : "Running"} + {controlled && taskStop.available ? ( + void taskStop.stop(task.taskId)} + > + {stopping ? "Stopping..." : "Stop"} + + ) : null}
{command ? ( @@ -67,7 +98,20 @@ export function BackgroundWorkDetails(props: {
) : null} - {itemId && streamsOutput && item?.status === "running" ? ( + {controlled && followsTasks ? ( +
+ {followedFile ? ( +
+ Following {followedFile} +
+ ) : null} + +
+ ) : itemId && streamsOutput && item?.status === "running" ? (
( - props.source - ? props.source(input) - : orchestrationEnvironment.v2.commandOutput({ environmentId: props.environmentId, input }), + return ( + + ); +} + +/** Follows a command the agent left running; mount only while its details are open. */ +export function BackgroundTaskOutputPanel(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly taskId: string; + readonly compact?: boolean; +}) { + // A task runs until it leaves the roster, so a quiet one says so rather than showing nothing. + return ( + No output yet.
} + /> ); +} + +function CommandOutputBody(props: { + readonly query: Parameters>[0]; + readonly onInitialOutputReady?: (() => void) | undefined; + readonly compact?: boolean | undefined; + readonly runningEmpty?: ReactNode; +}) { + const { data, error } = useEnvironmentQuery(props.query); useInitialOutputReady(data === null && !error, props.onInitialOutputReady); const scrollRef = useRef(null); const followRef = useRef(true); @@ -51,7 +85,11 @@ export function CommandOutputPanel(props: { if (text.length === 0) { if (error) return
Couldn't load output: {error}
; if (data === null) return
Loading output…
; - return data.running ? null :
No output.
; + return data.running ? ( + (props.runningEmpty ?? null) + ) : ( +
No output.
+ ); } // Parse after all hooks to preserve the text/theme compiler cache. const spans = terminalOutputSpans(text, TERMINAL_PALETTES[resolvedTheme]); @@ -63,7 +101,7 @@ export function CommandOutputPanel(props: { followRef.current = element.scrollHeight - element.scrollTop - element.clientHeight <= FOLLOW_SLACK_PX; }} - className="max-h-80 overflow-auto text-muted-foreground" + className={`${props.compact ? "max-h-48" : "max-h-80"} overflow-auto text-muted-foreground`} > {data?.output.truncated ?
Earlier output not shown
: null} {spans.map((span, index) => diff --git a/apps/web/src/components/chat/ThreadDetailsCard.tsx b/apps/web/src/components/chat/ThreadDetailsCard.tsx index 4e8a47a39791a..a5e03123c3e72 100644 --- a/apps/web/src/components/chat/ThreadDetailsCard.tsx +++ b/apps/web/src/components/chat/ThreadDetailsCard.tsx @@ -78,10 +78,10 @@ export function ThreadDetailsCard({ const measure = () => { const frame = element.closest("[data-thread-details-card]"); const next = element.offsetHeight + (frame ? frame.offsetHeight - frame.clientHeight : 0); - // PR lists, servers, lineage, side chats and automations scroll as they grow. Counting them + // PR lists, running work, servers, lineage, side chats and automations scroll as they grow. Counting them // toward density would hide the section and workspace controls when the user asks to see more. const lists = element.querySelectorAll( - "[data-thread-relationships-panel], [data-thread-side-chats-panel], [data-thread-details-pr-list], [data-thread-automations-panel], [data-thread-details-server-overflow]", + "[data-thread-relationships-panel], [data-thread-side-chats-panel], [data-thread-details-pr-list], [data-thread-automations-panel], [data-thread-details-server-overflow], [data-thread-details-running-overflow], [data-thread-details-running-output]", ); const fittingHeight = next - Array.from(lists).reduce((sum, list) => sum + list.offsetHeight, 0); diff --git a/apps/web/src/components/chat/ThreadDetailsPanel.tsx b/apps/web/src/components/chat/ThreadDetailsPanel.tsx index bb1546fd4b824..c2783253e15ef 100644 --- a/apps/web/src/components/chat/ThreadDetailsPanel.tsx +++ b/apps/web/src/components/chat/ThreadDetailsPanel.tsx @@ -23,6 +23,7 @@ import { ThreadDetailsSection } from "./ThreadDetailsSection"; import { ThreadAutomationsPanel } from "./ThreadAutomationsPanel"; import { ThreadRelationshipsPanel } from "./ThreadRelationshipsControl"; import { ThreadServersPanel } from "./ThreadServersPanel"; +import { ThreadRunningPanel, type ThreadRunningBackgroundWork } from "./ThreadRunningPanel"; import { ThreadDetailsContext, ThreadDetailsUsage, @@ -84,6 +85,7 @@ interface ThreadDetailsPanelProps extends Pick< input: NewProjectScriptInput, ) => Promise; onDeleteProjectScript: (scriptId: string) => Promise; + backgroundWork: ThreadRunningBackgroundWork | null; usage: ThreadDetailsUsageInput; } @@ -240,6 +242,14 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { ) : null} + {density === "full" && !props.draftId && visible.running ? ( + + ) : null} + {density === "full" && !props.draftId && visible.servers ? ( ) : null} diff --git a/apps/web/src/components/chat/ThreadRunningPanel.tsx b/apps/web/src/components/chat/ThreadRunningPanel.tsx new file mode 100644 index 0000000000000..e8eb97b7df484 --- /dev/null +++ b/apps/web/src/components/chat/ThreadRunningPanel.tsx @@ -0,0 +1,317 @@ +import { useAtomValue } from "@effect/atom-react"; +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import type { ThreadStructure } from "@t3tools/client-runtime/state/shell"; +import { presentPendingBackgroundWork } from "@t3tools/client-runtime/state/thread-execution"; +import type { + EnvironmentId, + OrchestrationV2PendingBackgroundTask, + ThreadId, +} from "@t3tools/contracts"; +import { ChevronRightIcon, LoaderCircleIcon, SquareIcon, TerminalIcon } from "lucide-react"; +import { useState, type ReactNode } from "react"; + +import { revealThreadTerminal } from "../../lib/revealThreadTerminal"; +import { cn } from "../../lib/utils"; +import { terminalEnvironment } from "../../state/terminal"; +import { useKnownTerminalSessions } from "../../state/terminalSessions"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { BackgroundWorkDetails, backgroundTaskLaunch } from "./BackgroundWorkDetails"; +import { ShellCommandBlock } from "./ShellCommandBlock"; +import { ThreadDetailsControl } from "./ThreadDetailsControl"; +import { ThreadDetailsSection } from "./ThreadDetailsSection"; +import { THREAD_DETAILS_PANEL_ICON_CLASS } from "./threadDetailsPanelStyles"; +import { BackgroundTaskOutputPanel } from "./CommandOutputPanel"; +import { useBackgroundTaskControl, useBackgroundTaskStop } from "./useBackgroundTaskStop"; + +const RUNNING_PAGE_COUNT = 4; + +export interface ThreadRunningBackgroundWork { + readonly tasks: ReadonlyArray; + readonly turnItems: ThreadStructure["turnItems"]; + readonly stopDisabled: boolean; + readonly stopping: boolean; + readonly onStop: () => Promise; + readonly onOpenThread: (threadId: ThreadId) => void; +} + +/** Commands the thread's agent left running and its terminals' foreground processes. Hidden while none run. */ +export function ThreadRunningPanel(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly backgroundWork: ThreadRunningBackgroundWork | null; +}) { + const sessions = useKnownTerminalSessions(props); + const canInterrupt = useAtomValue(terminalEnvironment.write.permissionAtom(props.environmentId)); + const writeTerminal = useAtomCommand(terminalEnvironment.write, "terminal interrupt"); + const taskStop = useBackgroundTaskStop(props.environmentId, props.threadId); + // Servers that stream task output show it in place; older ones keep the details dialog. + const followsTasks = useBackgroundTaskControl(props.environmentId); + const [expandedTaskId, setExpandedTaskId] = useState(null); + const work = props.backgroundWork; + // Subagents are listed under Lineage. + const tasks = (work?.tasks ?? []).filter((task) => task.kind !== "subagent"); + const labels = new Map( + presentPendingBackgroundWork(tasks)?.items.map((item) => [item.taskId, item.label]), + ); + const terminals = (sessions ?? []).flatMap(({ state: { summary } }) => + summary?.hasRunningSubprocess ? [summary] : [], + ); + const threadRef = scopeThreadRef(props.environmentId, props.threadId); + + const rows = [ + ...(work === null + ? [] + : tasks.map((task) => { + const name = labels.get(task.taskId) ?? task.kind; + // Only tasks the provider marks can be stopped alone or followed; others open the dialog. + const controlled = followsTasks && task.individualControl === true; + const stoppable = taskStop.available && task.individualControl === true; + const expanded = controlled && expandedTaskId === task.taskId; + const { command } = backgroundTaskLaunch(task, work.turnItems); + const followedFile = task.outputFile; + const label = ( + <> + + + {controlled ? ( + + ) : null} + + ); + return { + key: `agent:${task.taskId}`, + node: ( + void taskStop.stop(task.taskId)} + /> + ) : null + } + > + {controlled ? ( + setExpandedTaskId(expanded ? null : task.taskId)} + > + {label} + + ) : ( + + } + > + {label} + + )} + + ), + output: expanded ? ( +
+ {command ? ( +
+ +
+ ) : null} + {followedFile ? ( +
+ Following {followedFile} +
+ ) : null} + +
+ ) : null, + }; + })), + ...terminals.map(({ terminalId, label }) => { + const name = label || terminalId; + return { + key: `terminal:${terminalId}`, + output: null, + node: ( + + void writeTerminal({ + environmentId: props.environmentId, + input: { threadId: props.threadId, terminalId, data: "\u0003" }, + }) + } + /> + ) : null + } + > + revealThreadTerminal(threadRef, terminalId)} + > + + + + + ), + }; + }), + ]; + if (rows.length === 0) return null; + + return ( + !(taskStop.available && task.individualControl)) ? ( + + void work.onStop()} + /> + } + > + {work.stopping ? "Stopping..." : "Stop agent work"} + + + Stops everything the agent runs in the background, including subagents + + + ) : null + } + > +
    + {rows.map((row, index) => ( +
  • = RUNNING_PAGE_COUNT || undefined} + className="rounded-lg" + > + {row.node} + {row.output ? ( + // Open output scrolls inside its own bound and does not decide the card's density. +
    + {row.output} +
    + ) : null} +
    • + ))} +
+ + ); +} + +function RunningRow(props: { + readonly stop: ReactNode; + readonly stopBesideChevron?: boolean; + readonly children: ReactNode; +}) { + return ( +
+ {props.children} + {props.stop ? ( +
+ {props.stop} +
+ ) : null} +
+ ); +} + +function StopButton(props: { + readonly label: string; + readonly tooltip: string; + readonly pending: boolean; + readonly onClick: () => void; +}) { + return ( + + + } + > + {props.pending ? ( + + ) : ( + + )} + + {props.tooltip} + + ); +} + +function RowText(props: { + readonly title: string; + readonly detail: string; + readonly padEnd?: boolean; +}) { + return ( + + + {props.title} + + + {props.detail} + + + ); +} diff --git a/apps/web/src/components/chat/threadDetailsItems.ts b/apps/web/src/components/chat/threadDetailsItems.ts index e6f94cc4e11c8..2cf94e43a4ab6 100644 --- a/apps/web/src/components/chat/threadDetailsItems.ts +++ b/apps/web/src/components/chat/threadDetailsItems.ts @@ -8,6 +8,7 @@ export const THREAD_DETAILS_ITEMS = [ { id: "branch", label: "Branch and pull request", defaultVisible: true }, { id: "gitActions", label: "Git actions", defaultVisible: true }, { id: "changes", label: "Changes", defaultVisible: true }, + { id: "running", label: "Running commands", defaultVisible: true }, { id: "servers", label: "Servers", defaultVisible: true }, { id: "automations", label: "Automations", defaultVisible: true }, { id: "lineage", label: "Lineage", defaultVisible: true }, diff --git a/apps/web/src/components/chat/useBackgroundTaskStop.ts b/apps/web/src/components/chat/useBackgroundTaskStop.ts new file mode 100644 index 0000000000000..8249094f2e3c7 --- /dev/null +++ b/apps/web/src/components/chat/useBackgroundTaskStop.ts @@ -0,0 +1,62 @@ +import { useAtomValue } from "@effect/atom-react"; +import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; +import { useState } from "react"; + +import { useServerConfigs } from "../../state/entities"; +import { orchestrationEnvironment } from "../../state/orchestration"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { stackedThreadToast, toastManager } from "../ui/toast"; + +const STOP_RETRY_AFTER_MS = 10_000; + +/** Older servers answer neither per-task stop nor task output, so both stay hidden there. */ +export function useBackgroundTaskControl(environmentId: EnvironmentId): boolean { + return ( + useServerConfigs().get(environmentId)?.environment.capabilities.backgroundTaskControl === true + ); +} + +/** Stops one background task. A stopped task leaves the roster, which unmounts its row. */ +export function useBackgroundTaskStop(environmentId: EnvironmentId, threadId: ThreadId) { + const supported = useBackgroundTaskControl(environmentId); + const permitted = useAtomValue( + orchestrationEnvironment.stopBackgroundTask.permissionAtom(environmentId), + ); + const stopTask = useAtomCommand(orchestrationEnvironment.stopBackgroundTask, { + label: "background task stop", + }); + const [stopping, setStopping] = useState>(new Set()); + + const stop = async (taskId: string) => { + if (stopping.has(taskId)) return; + setStopping((current) => new Set(current).add(taskId)); + const result = await stopTask({ environmentId, input: { threadId, taskId } }); + const release = () => + setStopping((current) => { + const next = new Set(current); + next.delete(taskId); + return next; + }); + if (result._tag === "Success") { + // The row normally leaves first; if the provider never reports the stop, allow a retry. + window.setTimeout(release, STOP_RETRY_AFTER_MS); + return; + } + release(); + if (isAtomCommandInterrupted(result)) return; + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Could not stop background task", + description: error instanceof Error ? error.message : String(error), + }), + ); + }; + + return { available: supported && permitted, stopping, stop }; +} diff --git a/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx b/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx index a60d7f24b18cc..10fcbbc97fb0b 100644 --- a/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx +++ b/apps/web/src/components/chat/useThreadLifecycle.browser.test.tsx @@ -118,6 +118,7 @@ function RestoreBranchHarness(props: { runtimeMode: "full-access" as never, isWorking: false, pendingBackgroundTasks: [{ taskId: "background-command", kind: "command" }] as never, + runningBackgroundTasks: [], turnItems: [], activePendingApproval: null as never, activePendingUserInput: null as never, diff --git a/apps/web/src/components/chat/useThreadLifecycle.tsx b/apps/web/src/components/chat/useThreadLifecycle.tsx index ec2b323b9b1ee..306e372d8309f 100644 --- a/apps/web/src/components/chat/useThreadLifecycle.tsx +++ b/apps/web/src/components/chat/useThreadLifecycle.tsx @@ -106,6 +106,7 @@ interface ThreadLifecycleInput { readonly runtimeMode: RuntimeMode; readonly isWorking: boolean; readonly pendingBackgroundTasks: Parameters[0]["tasks"]; + readonly runningBackgroundTasks: Parameters[0]["tasks"]; readonly turnItems: Parameters[0]["turnItems"]; readonly activePendingApproval: PendingThreadRequests["activePendingApproval"]; readonly activePendingUserInput: PendingThreadRequests["activePendingUserInput"]; @@ -158,6 +159,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { runtimeMode, isWorking, pendingBackgroundTasks, + runningBackgroundTasks, turnItems, activePendingApproval, activePendingUserInput, @@ -606,6 +608,17 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { onOpenThread: onOpenRelatedThreadStable, }) : null; + const backgroundWork = activeThread + ? { + tasks: runningBackgroundTasks, + turnItems, + // While a turn runs, the composer's Stop owns the interrupt. + stopDisabled: isWorking || !canOperateThread || isStoppingBackgroundWork, + stopping: isStoppingBackgroundWork, + onStop: handleStopBackgroundWorkStable, + onOpenThread: onOpenRelatedThreadStable, + } + : null; // The relative times read the clock; nowMinute re-renders them on the minute. const snoozedUntil = activeThreadShell?.snoozedUntil ?? null; const settledAt = activeThreadShell?.settledAt ?? null; @@ -696,6 +709,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { shortcuts, acknowledgeActiveThreadWoke, activeBranchMismatchKey, + backgroundWork, // A Stop that went unanswered needs the user; one in progress is the thread's work. attentionNow: runStop?.state === "unresponsive" ? runStopNow : childInputNow, workNow: runStop?.state === "stopping" ? runStopNow : backgroundWorkNow, diff --git a/apps/web/src/components/settings/settingsSearch.ts b/apps/web/src/components/settings/settingsSearch.ts index 4f61ba41c0d86..254453650865e 100644 --- a/apps/web/src/components/settings/settingsSearch.ts +++ b/apps/web/src/components/settings/settingsSearch.ts @@ -348,7 +348,7 @@ export const SETTINGS_SEARCH_ITEMS = [ title: "Thread details", to: "/settings/appearance", searchTerms: [ - "customize show hide panel card environment editor scripts branch pull request git changes servers automations lineage usage limits 5 hour weekly quota tokens context window", + "customize show hide panel card environment editor scripts branch pull request git changes running commands background terminal processes servers automations lineage usage limits 5 hour weekly quota tokens context window", ], }, { diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 09ce63e2a3e65..f3073dcae8542 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -290,6 +290,14 @@ Select a server to open it in the browser panel. **Stop** ends the server's job and keeps the agent, the terminal, and T3 Code running. Stop needs terminal access to the environment and is not available on Windows. +**Running** in the thread details card lists commands the agent left running in +the background, such as a dev server, and terminals with a running process. +Select an agent command to follow its output, or a terminal to open it. When the +command sends its output to a log file, such as `> dev.log 2>&1`, that file is +followed instead. Stop on +an agent command ends only that command; on a terminal it sends Ctrl+C. Output +and single-command Stop are available for Claude. + To choose what the thread details card shows, right-click the card or open **Settings → Appearance → Thread details**. Hidden rows keep their keyboard shortcuts, such as Open in editor and project scripts. **Reset to default** brings diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index dd435d820d641..5dbce707282d8 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -54,6 +54,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof ORCHESTRATION_V2_WS_METHODS.subscribeShell | typeof ORCHESTRATION_V2_WS_METHODS.subscribeThread | typeof ORCHESTRATION_V2_WS_METHODS.subscribeCommandOutput + | typeof ORCHESTRATION_V2_WS_METHODS.subscribeBackgroundTaskOutput | typeof ORCHESTRATION_V2_WS_METHODS.subscribePromptSuggestion | typeof ORCHESTRATION_V2_WS_METHODS.searchThreadStream | typeof WS_METHODS.subscribeAuthAccess diff --git a/packages/client-runtime/src/state/commandPermissions.test.ts b/packages/client-runtime/src/state/commandPermissions.test.ts index d8da76a4b6906..953befc0192aa 100644 --- a/packages/client-runtime/src/state/commandPermissions.test.ts +++ b/packages/client-runtime/src/state/commandPermissions.test.ts @@ -204,23 +204,25 @@ describe("command permissions", () => { ), ); - it.effect.each([WS_METHODS.scheduledTasksDelete, ORCHESTRATION_V2_WS_METHODS.stopWorkflow])( - "uses the target grant for both availability and execution of %s", - (method) => - Effect.scoped( - Effect.gen(function* () { - const registry = yield* setup; - const permissions = createCommandPermissions(runtime, method); - registry.set(sessions(env), AsyncResult.success(grant(true))); - expect(registry.get(permissions.permissionAtom(env))).toBe(true); - expect(permissions.permissionAtom(env)).toBe(permissions.permissionAtom(env)); - yield* permissions.authorize(registry, env); - registry.set(sessions(other), AsyncResult.success(grant(false))); - expect(registry.get(permissions.permissionAtom(other))).toBe(false); - const denied = yield* permissions.authorize(registry, other).pipe(Effect.flip); - expect(denied._tag).toBe("EnvironmentAuthorizationError"); - }), - ), + it.effect.each([ + WS_METHODS.scheduledTasksDelete, + ORCHESTRATION_V2_WS_METHODS.stopWorkflow, + ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask, + ])("uses the target grant for both availability and execution of %s", (method) => + Effect.scoped( + Effect.gen(function* () { + const registry = yield* setup; + const permissions = createCommandPermissions(runtime, method); + registry.set(sessions(env), AsyncResult.success(grant(true))); + expect(registry.get(permissions.permissionAtom(env))).toBe(true); + expect(permissions.permissionAtom(env)).toBe(permissions.permissionAtom(env)); + yield* permissions.authorize(registry, env); + registry.set(sessions(other), AsyncResult.success(grant(false))); + expect(registry.get(permissions.permissionAtom(other))).toBe(false); + const denied = yield* permissions.authorize(registry, other).pipe(Effect.flip); + expect(denied._tag).toBe("EnvironmentAuthorizationError"); + }), + ), ); it.effect("waits for an initial grant", () => Effect.scoped( diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index 657a9cd62dab1..38ab327bb037e 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -173,6 +173,18 @@ export function createOrchestrationEnvironmentAtoms( Stream.filter((view) => view !== EMPTY_COMMAND_OUTPUT), ), }), + // A background command's output, only while its details are open. It never completes: + // the task's own roster entry, not this stream, says when it stops. + backgroundTaskOutput: createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:orchestration-v2:background-task-output", + tag: ORCHESTRATION_V2_WS_METHODS.subscribeBackgroundTaskOutput, + idleTtlMs: 0, + transform: (frames) => + frames.pipe( + Stream.scan(() => EMPTY_COMMAND_OUTPUT, applyCommandOutputFrame), + Stream.filter((view) => view !== EMPTY_COMMAND_OUTPUT), + ), + }), // The open thread's predicted next prompt, while its composer is mounted. promptSuggestion: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:orchestration-v2:prompt-suggestion", @@ -202,6 +214,10 @@ export function createOrchestrationEnvironmentAtoms( label: "environment-data:orchestration:stop-workflow", tag: ORCHESTRATION_V2_WS_METHODS.stopWorkflow, }), + stopBackgroundTask: createEnvironmentRpcCommand(runtime, { + label: "environment-data:orchestration:stop-background-task", + tag: ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask, + }), forceStopThread: createEnvironmentRpcCommand(runtime, { label: "environment-data:orchestration:force-stop-thread", tag: ORCHESTRATION_V2_WS_METHODS.forceStopThread, diff --git a/packages/contracts/src/clientRpcPermissions.ts b/packages/contracts/src/clientRpcPermissions.ts index 8b4a7d48d279c..7de51fdf83416 100644 --- a/packages/contracts/src/clientRpcPermissions.ts +++ b/packages/contracts/src/clientRpcPermissions.ts @@ -69,6 +69,7 @@ export const CLIENT_GUARDED_RPC_SCOPES = { [WS_METHODS.scheduledTasksRunNow]: AuthOrchestrationOperateScope, [WS_METHODS.scheduledTasksRotateWebhookToken]: AuthOrchestrationOperateScope, [ORCHESTRATION_V2_WS_METHODS.stopWorkflow]: AuthOrchestrationOperateScope, + [ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask]: AuthOrchestrationOperateScope, // Force stop kills agent processes, as Stop on a terminal does. [ORCHESTRATION_V2_WS_METHODS.forceStopThread]: AuthOrchestrationOperateScope, // A guest link hands thread access to someone outside, like a pairing link. diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 5e9cca70752b6..9ffe7fca49f1f 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -155,6 +155,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ desktopAppChannelSwitch: Schema.optionalKey(Schema.Boolean), /** Subscribe to command output only when advertised; older servers never send it. */ commandOutputStreaming: Schema.optionalKey(Schema.Boolean), + /** Serves orchestration.stopBackgroundTask and orchestration.subscribeBackgroundTaskOutput. */ + backgroundTaskControl: Schema.optionalKey(Schema.Boolean), /** Hosts headless Chromium tabs streamed over /api/preview-stream for clients without a local runtime. */ serverBrowser: Schema.optionalKey(Schema.Boolean), /** Server links to other environments (`PeerLink`) and its agents can work diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 9c66612458187..74b5acd5dcf30 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1040,6 +1040,10 @@ const PendingBackgroundTaskFields = { * subagent. Only the agent knows whether a command is a wait or a dev server it left running. */ awaited: Schema.optional(Schema.Literal(true)), + /** The provider can stop this task alone and stream its output. */ + individualControl: Schema.optional(Schema.Literal(true)), + /** The file the task's command sends its output to, which its output stream follows. */ + outputFile: Schema.optional(TrimmedNonEmptyString), }; export const OrchestrationV2PendingBackgroundTask = kindUnionWithFallback( @@ -3648,12 +3652,14 @@ export const ORCHESTRATION_V2_WS_METHODS = { getWorkflowScript: "orchestration.getWorkflowScript", getNativeChildWork: "orchestration.getNativeChildWork", stopWorkflow: "orchestration.stopWorkflow", + stopBackgroundTask: "orchestration.stopBackgroundTask", forceStopThread: "orchestration.forceStopThread", getTurnItem: "orchestration.getTurnItem", launchThread: "orchestration.launchThread", subscribeShell: "orchestration.subscribeShell", subscribeThread: "orchestration.subscribeThread", subscribeCommandOutput: "orchestration.subscribeCommandOutput", + subscribeBackgroundTaskOutput: "orchestration.subscribeBackgroundTaskOutput", subscribePromptSuggestion: "orchestration.subscribePromptSuggestion", } as const; @@ -3684,6 +3690,14 @@ export class OrchestrationV2CommandOutputError extends Schema.TaggedError()( + "OrchestrationV2StopBackgroundTaskError", + { + threadId: ThreadId, + taskId: Schema.String, + reason: Schema.Literals(["not-running", "unsupported", "unavailable", "stop-failed"]), + cause: Schema.optional(Schema.Defect()), + }, +) { + override get message(): string { + switch (this.reason) { + case "not-running": + return "This background task is no longer running."; + case "unsupported": + return "This provider cannot stop a single background task."; + case "unavailable": + return "This background task has no live provider session."; + case "stop-failed": + return "Failed to stop the background task."; + } + } +} + export const OrchestrationV2ForceStopThreadInput = Schema.Struct({ threadId: ThreadId, /** The run the user saw ignore Stop; Force stop refuses once a newer run has started. */ @@ -4341,6 +4385,10 @@ export const OrchestrationV2RpcSchemas = { input: OrchestrationV2StopWorkflowInput, output: Schema.Void, }, + stopBackgroundTask: { + input: OrchestrationV2StopBackgroundTaskInput, + output: Schema.Void, + }, forceStopThread: { input: OrchestrationV2ForceStopThreadInput, output: OrchestrationV2ForceStopThreadResult, @@ -4373,6 +4421,10 @@ export const OrchestrationV2RpcSchemas = { input: OrchestrationV2SubscribeCommandOutputInput, output: OrchestrationV2CommandOutputFrame, }, + subscribeBackgroundTaskOutput: { + input: OrchestrationV2SubscribeBackgroundTaskOutputInput, + output: OrchestrationV2CommandOutputFrame, + }, subscribePromptSuggestion: { input: OrchestrationV2SubscribePromptSuggestionInput, output: OrchestrationV2PromptSuggestion, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 0f6af19188e18..a4a4bf7fb9ccd 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -226,6 +226,7 @@ import { OrchestrationGetWorkflowScriptError, OrchestrationV2GetNativeChildWorkError, OrchestrationV2StopWorkflowError, + OrchestrationV2StopBackgroundTaskError, OrchestrationV2ForceStopThreadError, OrchestrationV2DispatchCommandError, OrchestrationV2GetShellSnapshotError, @@ -1875,6 +1876,15 @@ const WsOrchestrationV2StopWorkflowRpc = Rpc.make(ORCHESTRATION_V2_WS_METHODS.st error: Schema.Union([OrchestrationV2StopWorkflowError, EnvironmentAuthorizationError]), }); +const WsOrchestrationV2StopBackgroundTaskRpc = Rpc.make( + ORCHESTRATION_V2_WS_METHODS.stopBackgroundTask, + { + payload: OrchestrationV2RpcSchemas.stopBackgroundTask.input, + success: OrchestrationV2RpcSchemas.stopBackgroundTask.output, + error: Schema.Union([OrchestrationV2StopBackgroundTaskError, EnvironmentAuthorizationError]), + }, +); + const WsOrchestrationV2ForceStopThreadRpc = Rpc.make(ORCHESTRATION_V2_WS_METHODS.forceStopThread, { payload: OrchestrationV2RpcSchemas.forceStopThread.input, success: OrchestrationV2RpcSchemas.forceStopThread.output, @@ -1935,6 +1945,16 @@ const WsOrchestrationV2SubscribeCommandOutputRpc = Rpc.make( }, ); +const WsOrchestrationV2SubscribeBackgroundTaskOutputRpc = Rpc.make( + ORCHESTRATION_V2_WS_METHODS.subscribeBackgroundTaskOutput, + { + payload: OrchestrationV2RpcSchemas.subscribeBackgroundTaskOutput.input, + success: OrchestrationV2RpcSchemas.subscribeBackgroundTaskOutput.output, + error: Schema.Union([OrchestrationV2CommandOutputError, EnvironmentAuthorizationError]), + stream: true, + }, +); + const WsOrchestrationV2SubscribePromptSuggestionRpc = Rpc.make( ORCHESTRATION_V2_WS_METHODS.subscribePromptSuggestion, { @@ -2264,6 +2284,7 @@ export const WsRpcGroup = RpcGroup.make( WsOrchestrationV2GetWorkflowScriptRpc, WsOrchestrationV2GetNativeChildWorkRpc, WsOrchestrationV2StopWorkflowRpc, + WsOrchestrationV2StopBackgroundTaskRpc, WsOrchestrationV2ForceStopThreadRpc, WsOrchestrationV2GetTurnItemRpc, WsOrchestrationV2GetTurnDiffRpc, @@ -2282,5 +2303,6 @@ export const WsRpcGroup = RpcGroup.make( WsOrchestrationV2SubscribeShellRpc, WsOrchestrationV2SubscribeThreadRpc, WsOrchestrationV2SubscribeCommandOutputRpc, + WsOrchestrationV2SubscribeBackgroundTaskOutputRpc, WsOrchestrationV2SubscribePromptSuggestionRpc, ).middleware(RpcScopeAuthorization); diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 98f71a7473a84..1c8261c85661a 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -689,6 +689,16 @@ export interface ProviderAdapterV2SessionRuntime { readonly taskIds: ReadonlyArray; readonly awaited: boolean; }) => Effect.Effect<{ readonly matched: ReadonlyArray }, ProviderAdapterV2Error>; + /** Stops one task from the provider thread's `pendingBackgroundTasks`, leaving the rest running. */ + readonly stopBackgroundTask?: (input: { + readonly providerThread: OrchestrationV2ProviderThread; + readonly taskId: string; + }) => Effect.Effect; + /** A background task's raw output, from its recent tail onward, read only while consumed. */ + readonly backgroundTaskOutput?: (input: { + readonly providerThread: OrchestrationV2ProviderThread; + readonly taskId: string; + }) => Stream.Stream; readonly interruptTurn: ( input: ProviderAdapterV2InterruptInput, ) => Effect.Effect; diff --git a/packages/shared/package.json b/packages/shared/package.json index f5f769236f906..4a2475f92f964 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -99,6 +99,10 @@ "types": "./src/shell.ts", "import": "./src/shell.ts" }, + "./shellStdoutFile": { + "types": "./src/shellStdoutFile.ts", + "import": "./src/shellStdoutFile.ts" + }, "./semver": { "types": "./src/semver.ts", "import": "./src/semver.ts" diff --git a/packages/shared/src/shellStdoutFile.test.ts b/packages/shared/src/shellStdoutFile.test.ts new file mode 100644 index 0000000000000..d5cf03f289f27 --- /dev/null +++ b/packages/shared/src/shellStdoutFile.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { shellStdoutFile } from "./shellStdoutFile.ts"; + +describe("shellStdoutFile", () => { + it.each([ + ["vp run dev", undefined], + ["vp run dev > dev.log 2>&1", "dev.log"], + ["vp run dev >> dev.log", "dev.log"], + ["vp run dev 1>dev.log", "dev.log"], + ["vp run dev &> dev.log", "dev.log"], + ["vp run dev > dev.log 2>&1 &", "dev.log"], + ["vp run dev 2>&1 | tee -a logs/dev.log", "logs/dev.log"], + ["vp run dev |& tee dev.log", "dev.log"], + ["echo foo2>file.txt", "file.txt"], + ["PORT=$(get-port) vp run dev > dev.log", "dev.log"], + [ + `cd /c/repo; vp run dev > "C:/Users/me/AppData/Local/Temp/dev2.log" 2>&1`, + "C:/Users/me/AppData/Local/Temp/dev2.log", + ], + ["vp run dev > 'my logs/dev.log'", "my logs/dev.log"], + [String.raw`vp run dev > "C:\Users\me\dev.log"`, String.raw`C:\Users\me\dev.log`], + ["vp run dev > first.log; vp run dev > second.log", "second.log"], + ["vp run dev > dev.log # > other.log", "dev.log"], + // Only the last command's stdout counts: earlier commands finished before it started. + ["npm run build > build.log && npm run dev", undefined], + ["echo starting > status.txt; npm run dev", undefined], + ["cat > run.sh <<'EOF'\necho hi > inner.log\nEOF\nbash run.sh", undefined], + // stderr alone, descriptor copies, devices, and a bare tee leave stdout on the terminal. + ["vp run dev 2> errors.log", undefined], + ["vp run dev >&2", undefined], + ["vp run dev > dev.log >&2", undefined], + ["vp run dev > /dev/null 2>&1", undefined], + ["curl example.com > /dev/stderr", undefined], + ["vp run dev | tee -a", undefined], + // Redirect characters inside quotes are arguments, not redirects. + [`echo "a > b" && node server.js`, undefined], + [`grep '>' notes.txt`, undefined], + // Only the shell knows these paths and directories. + ["vp run dev > $LOG_FILE", undefined], + [`vp run dev > "$TMPDIR/dev.log"`, undefined], + ["vp run dev > ~/dev.log", undefined], + ["cd $APP && vp run dev > dev.log", undefined], + ["(cd web && npm run dev) > dev.log", undefined], + ])("%s", (command, expected) => { + expect(shellStdoutFile(command)?.path).toBe(expected); + }); + + it("keeps the directories the command changes into before writing", () => { + expect(shellStdoutFile("cd apps/web && cd .. && vp run dev > dev.log 2>&1 &")).toEqual({ + path: "dev.log", + cwdChanges: ["apps/web", ".."], + }); + }); +}); diff --git a/packages/shared/src/shellStdoutFile.ts b/packages/shared/src/shellStdoutFile.ts new file mode 100644 index 0000000000000..d9be53fa1d97c --- /dev/null +++ b/packages/shared/src/shellStdoutFile.ts @@ -0,0 +1,215 @@ +const WORD_END = new Set([" ", "\t", "\n", ";", "&", "|", "<", ">", "(", ")"]); + +type Token = + | { readonly kind: "word"; readonly value: string | null } + | { readonly kind: "separator" } + | { readonly kind: "pipe" } + // A null target is stdout duplicated to a descriptor or a process, or a path only the shell knows. + | { readonly kind: "redirect"; readonly stdout: boolean; readonly target: string | null }; + +function skipQuoted(command: string, start: number): number { + const quote = command[start]; + let index = start + 1; + while (index < command.length && command[index] !== quote) { + index += quote === '"' && command[index] === "\\" ? 2 : 1; + } + return index + 1; +} + +function skipBalanced(command: string, open: number): number { + const close = command[open] === "(" ? ")" : "}"; + let depth = 0; + let index = open; + while (index < command.length) { + const char = command[index]!; + if (char === "'" || char === '"' || char === "`") { + index = skipQuoted(command, index); + continue; + } + if (char === "\\") index += 1; + else if (char === command[open]) depth += 1; + else if (char === close && --depth === 0) return index + 1; + index += 1; + } + return index; +} + +/** Reads one shell word, joining quoted parts. Null when it needs expansion we cannot do. */ +function readWord(command: string, start: number): { value: string | null; end: number } { + let index = start; + while (command[index] === " " || command[index] === "\t") index += 1; + let value = ""; + let expands = false; + while (index < command.length && !WORD_END.has(command[index]!)) { + const char = command[index]!; + if (char === "'" || char === '"') { + const end = skipQuoted(command, index); + const inner = command.slice(index + 1, end - 1); + if (char === '"' && /[$`]/.test(inner)) expands = true; + // Inside double quotes a backslash escapes only these, so Windows paths keep theirs. + value += char === '"' ? inner.replace(/\\([$`"\\\n])/g, "$1") : inner; + index = end; + } else if (char === "\\") { + if (command[index + 1] !== "\n") value += command[index + 1] ?? ""; + index += 2; + } else if (char === "$" && (command[index + 1] === "(" || command[index + 1] === "{")) { + index = skipBalanced(command, index + 1); + expands = true; + } else if (char === "`") { + index = skipQuoted(command, index); + expands = true; + } else { + if ("$*?[{".includes(char) || (value === "" && char === "~")) expands = true; + value += char; + index += 1; + } + } + return { value: value === "" || expands ? null : value, end: index }; +} + +/** Null for constructs this scanner does not model, such as subshells. */ +function tokenize(command: string): ReadonlyArray | null { + const tokens: Array = []; + const heredocs: Array<{ readonly delimiter: string; readonly stripTabs: boolean }> = []; + let index = 0; + const redirect = (descriptor: number | "&", operatorEnd: number) => { + const stdout = descriptor === 1 || descriptor === "&"; + if (command[operatorEnd] === "&" || command[operatorEnd] === "(") { + // `>&2` and `>(...)` send output to a descriptor or a process, not a file. + const end = + command[operatorEnd] === "(" + ? skipBalanced(command, operatorEnd) + : readWord(command, operatorEnd + 1).end; + tokens.push({ kind: "redirect", stdout, target: null }); + return end; + } + const word = readWord(command, operatorEnd); + tokens.push({ kind: "redirect", stdout, target: word.value }); + return Math.max(word.end, operatorEnd); + }; + while (index < command.length) { + const char = command[index]!; + const next = command[index + 1]; + if (char === " " || char === "\t") { + index += 1; + } else if (char === "\\" && next === "\n") { + index += 2; + } else if (char === "\n" || char === ";") { + tokens.push({ kind: "separator" }); + index += 1; + if (char === "\n") { + for (const heredoc of heredocs.splice(0)) { + while (index < command.length) { + const lineEnd = command.indexOf("\n", index); + const end = lineEnd === -1 ? command.length : lineEnd; + const line = command.slice(index, end); + index = end + 1; + if ((heredoc.stripTabs ? line.replace(/^\t+/, "") : line) === heredoc.delimiter) break; + } + } + } + } else if (char === "#") { + const lineEnd = command.indexOf("\n", index); + index = lineEnd === -1 ? command.length : lineEnd; + } else if (char === "&") { + if (next === ">") { + index = redirect("&", index + (command[index + 2] === ">" ? 3 : 2)); + } else { + tokens.push({ kind: "separator" }); + index += next === "&" ? 2 : 1; + } + } else if (char === "|") { + tokens.push(next === "|" ? { kind: "separator" } : { kind: "pipe" }); + index += next === "|" || next === "&" ? 2 : 1; + } else if (char === "(" || char === ")") { + return null; + } else if (char === ">" || (/\d/.test(char) && /^\d+[<>]/.test(command.slice(index)))) { + const digits = /^\d*/.exec(command.slice(index))![0]; + const operator = index + digits.length; + const descriptor = digits === "" ? 1 : Number(digits); + if (command[operator] === "<") { + index = readWord(command, operator + 1).end; + } else { + const append = command[operator + 1] === ">" || command[operator + 1] === "|"; + index = redirect(descriptor, operator + (append ? 2 : 1)); + } + } else if (char === "<") { + if (next === "<" && command[index + 2] === "<") { + index = readWord(command, index + 3).end; + } else if (next === "<") { + const stripTabs = command[index + 2] === "-"; + const start = index + (stripTabs ? 3 : 2); + const word = readWord(command, start); + const raw = command + .slice(start, word.end) + .trim() + .replace(/^['"]|['"]$/g, ""); + heredocs.push({ delimiter: word.value ?? raw, stripTabs }); + index = word.end; + } else if (next === "(") { + index = skipBalanced(command, index + 1); + } else { + index = readWord(command, next === "&" ? index + 2 : index + 1).end; + } + } else { + const word = readWord(command, index); + tokens.push({ kind: "word", value: word.value }); + index = Math.max(word.end, index + 1); + } + } + return tokens; +} + +export interface ShellStdoutFile { + /** As written in the command; relative paths resolve after `cwdChanges`. */ + readonly path: string; + /** Directories the command `cd`s into, in order, before the command that writes the file. */ + readonly cwdChanges: ReadonlyArray; +} + +/** + * The file the last command in a shell command line sends its stdout to: its last `>`, `>>`, + * `1>`, `&>` or `| tee` target. Undefined when that command's stdout stays on the terminal, goes + * to a device, or names a path only the shell could work out. + */ +export function shellStdoutFile(command: string): ShellStdoutFile | undefined { + const tokens = tokenize(command); + if (tokens === null) return undefined; + const commands: Array> = [[]]; + for (const token of tokens) { + if (token.kind === "separator") commands.push([]); + else commands.at(-1)!.push(token); + } + const filled = commands.filter((tokens) => tokens.length > 0); + const last = filled.at(-1); + if (last === undefined) return undefined; + + const cwdChanges: Array = []; + for (const earlier of filled.slice(0, -1)) { + const words = earlier.filter((token) => token.kind === "word"); + if (words[0]?.value !== "cd") continue; + const directory = words[1]?.value; + if (words.length !== 2 || directory == null || directory === "-") return undefined; + cwdChanges.push(directory); + } + + let target: string | null | undefined; + let stage: Array = []; + let teeWrote = false; + for (const token of last) { + if (token.kind === "pipe") { + stage = []; + teeWrote = false; + } else if (token.kind === "redirect") { + if (token.stdout) target = token.target; + } else if (token.kind === "word") { + stage.push(token.value); + if (stage[0] === "tee" && stage.length > 1 && !teeWrote && !token.value?.startsWith("-")) { + target = token.value; + teeWrote = true; + } + } + } + if (target == null || target.startsWith("/dev/")) return undefined; + return { path: target, cwdChanges }; +}