diff --git a/src/servers/api/chat/chat.ts b/src/servers/api/chat/chat.ts index 0e6d00f2..e3b8e3a5 100644 --- a/src/servers/api/chat/chat.ts +++ b/src/servers/api/chat/chat.ts @@ -1,4 +1,6 @@ import type { Context } from 'hono'; +import { emitToSession } from './websocket'; +import { pauseBackgroundTask, restartBackgroundTask } from './task-restart'; import { createRouter } from '../../create-router'; import * as errors from '@@/custom-errors'; import * as sidecar from '@@/sidecar-registry'; @@ -281,6 +283,49 @@ chatRouter.delete('/tasks/:id', async (ctx) => { return ctx.json(result, result.stopped ? 200 : 400); }); +// POST /chat/tasks/:id/restart — run this job's command again, without involving the agent. +// +// The platform supervises what it respawns, because the harness stopped watching when its own child died. +// It emits the same two events a harness-owned task emits, so the tray, the log view and reconnects need to +// know nothing about the difference. `sessionId` comes from the client because that is who knows which +// conversation the chip is being looked at in. +chatRouter.post('/tasks/:id/restart', async (ctx) => { + const who = await chatIdentity(ctx.get('user')); + const taskId = ctx.req.param('id'); + const body = await ctx.req.json<{ sessionId?: string }>().catch(() => ({}) as { sessionId?: string }); + + const result = await restartBackgroundTask(who, taskId, (code: number) => { + if (!body.sessionId) return; + emitToSession(body.sessionId, { + type: 'task:notification', + taskId, + status: code === 0 ? 'completed' : 'failed', + summary: code === 0 ? 'Restarted job finished' : `Restarted job exited with code ${code}`, + }); + }); + + // Announced only on success, and only after the process exists — a `task:started` for something that + // failed to spawn would leave a chip running forever with nothing behind it. + if (result.ok && body.sessionId) { + emitToSession(body.sessionId, { + type: 'task:started', + taskId, + description: result.command?.split('\n')[0]?.slice(0, 80) ?? 'restarted job', + taskType: 'local_bash', + }); + } + + return ctx.json(result, result.ok ? 200 : 400); +}); + +// POST /chat/tasks/:id/pause — freeze or thaw a running job, keeping everything it holds. +chatRouter.post('/tasks/:id/pause', async (ctx) => { + const who = await chatIdentity(ctx.get('user')); + const body = await ctx.req.json<{ paused?: boolean }>().catch(() => ({}) as { paused?: boolean }); + const result = await pauseBackgroundTask(who, ctx.req.param('id'), body.paused !== false); + return ctx.json(result, result.ok ? 200 : 400); +}); + // GET /chat/models — Claude tiers only (the runner is the `claude` CLI). chatRouter.get('/models', async (ctx: Context) => { const who = await chatIdentity(ctx.get('user')); diff --git a/src/servers/api/chat/claude-sessions.ts b/src/servers/api/chat/claude-sessions.ts index 9ea85634..9bbb8891 100644 --- a/src/servers/api/chat/claude-sessions.ts +++ b/src/servers/api/chat/claude-sessions.ts @@ -711,7 +711,7 @@ export type BackgroundTaskDetail = * `///subagents/agent-.jsonl` (4). `taskId` is `TASK_ID_RE`-checked above, so * it carries no glob or shell metacharacter into either. */ -function findTaskOutput(who: ChatIdentity, taskId: string): string | null { +export function findTaskOutput(who: ChatIdentity, taskId: string): string | null { // The member's `claude` runs under `--reset-env`, so its TMPDIR is the passwd default, not ours. Both // resolve to /tmp on Linux; a server started with a TMPDIR of its own would look in the wrong place. const tmpRoot = process.env.TMPDIR ?? '/tmp'; @@ -733,6 +733,23 @@ function tailFile(osUser: AsUser, filePath: string, bytes: number): { text: stri return { text: truncated ? text.slice(text.indexOf('\n') + 1) : text, truncated }; } +/** + * The pids with this file open — which is how a job with no recorded pid is found. + * + * `fuser` reads `/proc//fd`, which only the process's own user and root may open, so it runs AS the + * job's owner. Run as the platform it would find nothing for a member's job and report it already stopped. + */ +export async function pidsHoldingFile(who: ChatIdentity, path: string): Promise { + const argv = who.osUser ? runAsArgv(who.osUser, ['fuser', path]) : ['fuser', path]; + const proc = Bun.spawn(argv, { stdout: 'pipe', stderr: 'ignore' }); + const out = await new Response(proc.stdout).text(); + await proc.exited; + return out + .split(/\s+/) + .map((token) => Number(token.replace(/\D/g, ''))) + .filter((n) => Number.isInteger(n) && n > 1); +} + /** * Stop a running background job. * @@ -762,15 +779,7 @@ export async function stopBackgroundTask( // stopped. And the kill would then be cross-user: EPERM, silently swallowed by the catch below. const asUser = (command: string[]) => (who.osUser ? runAsArgv(who.osUser, command) : command); - const proc = Bun.spawn(asUser(['fuser', found]), { stdout: 'pipe', stderr: 'ignore' }); - const out = await new Response(proc.stdout).text(); - await proc.exited; - - const pids = out - .split(/\s+/) - .map((t) => Number(t.replace(/\D/g, ''))) - .filter((n) => Number.isInteger(n) && n > 1); - + const pids = await pidsHoldingFile(who, found); if (pids.length === 0) return { stopped: false, reason: 'already stopped' }; if (who.osUser) { diff --git a/src/servers/api/chat/task-restart.ts b/src/servers/api/chat/task-restart.ts new file mode 100644 index 00000000..cdc32fc6 --- /dev/null +++ b/src/servers/api/chat/task-restart.ts @@ -0,0 +1,202 @@ +import { openSync } from 'node:fs'; +import { basename, dirname, join } from 'node:path'; +import { runAsArgv } from '../../os-user'; +import { readTextAs } from '../../read-as-user'; +import { findTaskOutput, pidsHoldingFile, type ChatIdentity } from './claude-sessions'; + +/** + * Run a background job's command again, without asking the agent. + * + * ── What this owns, and what it does not ── + * + * A background task normally belongs entirely to the harness: it spawns the process, watches it, and emits + * `task:started` and `task:notification`. Everything downstream — the tray, the log view, reconnects — is + * built on exactly those two events. + * + * A restart cannot be that. The agent is not involved, so nothing would emit either event, and the harness + * stops watching the moment its own child dies. So the platform takes over supervision of the process it + * spawned: it emits the same two events for it, and watches for the exit itself. + * + * That is a second owner for the same kind of thing, and it is a real cost — accepted deliberately, so + * that restarting a dev server is a button rather than a sentence typed at an agent that may be mid-turn. + * The seam is contained here: nothing else learns that a task can have two kinds of parent. + * + * ── The output file is the anchor ── + * + * The respawned process appends to the SAME `.output` the harness created. That is what keeps + * everything else working untouched: the log view reads that file, and Stop finds the process by asking + * which pids hold it open. A new file would have meant teaching both about a second location. + */ + +/** In-flight restarts, so a second click stops the first rather than racing it. */ +const restarted = new Map(); + +export type RestartResult = { ok: boolean; command?: string; reason?: string }; + +/** + * The command that started a task, recovered by joining the transcript back to itself. + * + * The harness records the tool call (`tool_use`, carrying `command`) and, separately, its result — which + * is the text `Command running in background with ID: `. Neither alone is enough; together they + * are a join on `tool_use_id`, and that is the only place the command survives. + * + * The transcript is found from the OUTPUT PATH rather than by searching: a task file lives at + * `/claude-///tasks/.output`, so the slug and session name the + * transcript exactly. Guessing among a member's transcripts would be both slower and occasionally wrong. + */ +export async function commandForTask(who: ChatIdentity, taskId: string): Promise { + const output = findTaskOutput(who, taskId); + if (!output || !output.endsWith('.output')) return null; + + const sessionDir = dirname(dirname(output)); + const transcript = join( + who.home, + '.claude', + 'projects', + basename(dirname(sessionDir)), + `${basename(sessionDir)}.jsonl`, + ); + + let text: string; + try { + text = readTextAs(who.osUser, transcript); + } catch { + return null; + } + + const commands = new Map(); + let toolUseId: string | null = null; + + for (const line of text.split('\n')) { + if (!line) continue; + let entry: unknown; + try { + entry = JSON.parse(line); + } catch { + continue; + } + const content = (entry as { message?: { content?: unknown } })?.message?.content; + if (!Array.isArray(content)) continue; + + for (const block of content as Record[]) { + if (block.type === 'tool_use' && block.name === 'Bash') { + const command = (block.input as { command?: unknown } | undefined)?.command; + if (typeof block.id === 'string' && typeof command === 'string') commands.set(block.id, command); + } + if (block.type === 'tool_result') { + const body = typeof block.content === 'string' ? block.content : JSON.stringify(block.content ?? ''); + // Both phrasings the harness produces carry the id; this is the one that names a shell job. + if (body.includes(`background with ID: ${taskId}`) && typeof block.tool_use_id === 'string') { + toolUseId = block.tool_use_id; + } + } + } + } + + return toolUseId ? (commands.get(toolUseId) ?? null) : null; +} + +/** + * Stop whatever is running for this task and start its command again. + * + * `onExit` is how the caller emits `task:notification` — this module deliberately knows nothing about + * websockets or sessions. + */ +export async function restartBackgroundTask( + who: ChatIdentity, + taskId: string, + onExit: (code: number) => void, +): Promise { + const output = findTaskOutput(who, taskId); + if (!output) return { ok: false, reason: 'no such task' }; + + const command = await commandForTask(who, taskId); + if (!command) { + // Honest about which half failed: the file is there, the command is not recoverable. That happens when + // the transcript has been pruned, and no amount of retrying will change it. + return { ok: false, reason: 'cannot tell what command started this job — its transcript is gone' }; + } + + // Ours first, then anything else holding the file. A second click must replace the first restart, not + // leave two dev servers fighting over a port. + const previous = restarted.get(taskId); + if (previous) { + try { + previous.proc.kill(); + } catch { + /* already dead */ + } + restarted.delete(taskId); + } + + for (const pid of await pidsHoldingFile(who, output)) { + const argv = who.osUser ? runAsArgv(who.osUser, ['kill', '-TERM', String(pid)]) : null; + try { + if (argv) Bun.spawn(argv, { stdout: 'ignore', stderr: 'ignore' }); + else process.kill(pid, 'SIGTERM'); + } catch { + /* gone between listing and killing */ + } + } + // A dev server holding a port needs a moment to release it, and the whole point of a restart is that the + // new one binds. Not a fix for a process that ignores SIGTERM — that is what the second click is for. + await Bun.sleep(300); + + // Appended, never truncated: the log view is reading this file, and a restart is a new chapter in a job's + // life rather than a new job. + const fd = openSync(output, 'a'); + const argv = who.osUser ? runAsArgv(who.osUser, ['sh', '-c', command]) : ['sh', '-c', command]; + const proc = Bun.spawn(argv, { stdout: fd, stderr: fd, stdin: 'ignore' }); + restarted.set(taskId, { proc }); + + void proc.exited.then((code) => { + // Only report the exit of the process we are still tracking — a later restart owns the task by then. + if (restarted.get(taskId)?.proc === proc) { + restarted.delete(taskId); + onExit(code ?? 0); + } + }); + + return { ok: true, command }; +} + +/** Whether the platform (rather than the harness) is currently supervising this task. */ +export const isRestarted = (taskId: string): boolean => restarted.has(taskId); + +/** + * Freeze or thaw a job, without ending it. + * + * `SIGSTOP` and `SIGCONT`, which is the only pair that means "stop doing things and keep everything" — the + * process keeps its memory, its open files and its bound ports, and picks up mid-instruction. + * + * Worth knowing before pressing it: a paused server does not REFUSE connections, it stops answering them. + * A browser pointed at it hangs rather than erroring, and an HMR client may give up and reconnect on + * resume. That is inherent to what pausing is, not a shortcoming of this implementation. + * + * Signals go to every pid holding the output file — for a shell job that is the shell and whatever it + * spawned, and freezing only one of them would be worse than not freezing at all. + */ +export async function pauseBackgroundTask( + who: ChatIdentity, + taskId: string, + paused: boolean, +): Promise<{ ok: boolean; reason?: string }> { + const output = findTaskOutput(who, taskId); + if (!output) return { ok: false, reason: 'no such task' }; + + const pids = await pidsHoldingFile(who, output); + if (pids.length === 0) return { ok: false, reason: 'nothing is running for this job' }; + + const signal = paused ? '-STOP' : '-CONT'; + for (const pid of pids) { + const argv = who.osUser ? runAsArgv(who.osUser, ['kill', signal, String(pid)]) : null; + try { + if (argv) Bun.spawn(argv, { stdout: 'ignore', stderr: 'ignore' }); + else process.kill(pid, paused ? 'SIGSTOP' : 'SIGCONT'); + } catch { + // Gone between listing and signalling. Nothing to do, and nothing to report: the caller asked for a + // state, and "not running" satisfies "paused" closely enough to not be worth an error. + } + } + return { ok: true }; +} diff --git a/src/servers/api/chat/websocket.ts b/src/servers/api/chat/websocket.ts index b8dee245..c5da6ef5 100644 --- a/src/servers/api/chat/websocket.ts +++ b/src/servers/api/chat/websocket.ts @@ -318,6 +318,25 @@ function createMessageHandler(sessionId: string, model: string) { }; } +/** + * Push an event into a live session from OUTSIDE a turn. + * + * The restart button is the only caller: the platform respawns a process the harness is no longer watching, + * so nothing else would emit the `task:started` / `task:notification` pair that the tray, the log view and + * every reconnect are built on. Folded into the session first and then sent, in that order, so a browser + * that reconnects a second later sees the same thing as one that was already listening. + * + * Returns false when the session is not live — a restart is still legitimate then, it simply has nobody to + * tell. + */ +export function emitToSession(sessionId: string, msg: TurnMessage): boolean { + const session = sessionManager.getSession(sessionId); + if (!session) return false; + foldIntoSession(session, msg, session.meta.model ?? ''); + for (const socket of session.sockets) sendToClient(socket as ServerWebSocket, msg); + return true; +} + async function handleChat( ws: ServerWebSocket, msg: { diff --git a/src/workspaces/officerdev/src/apps/Chat/components/BackgroundTaskTray.tsx b/src/workspaces/officerdev/src/apps/Chat/components/BackgroundTaskTray.tsx index 8b083b81..88c1ae60 100644 --- a/src/workspaces/officerdev/src/apps/Chat/components/BackgroundTaskTray.tsx +++ b/src/workspaces/officerdev/src/apps/Chat/components/BackgroundTaskTray.tsx @@ -2,7 +2,7 @@ import type { ReactNode } from 'react'; import { useEffect, useRef, useState } from 'react'; import { toast } from 'sonner'; import { useClient } from 'hooks/useClient'; -import { Check, ChevronDown, CircleSlash, Loader2, Pin, Square, X } from 'lucide-react'; +import { Check, ChevronDown, CircleSlash, Loader2, Pause, Pin, Play, RotateCw, Square, X } from 'lucide-react'; import type { Tone } from '@/components/Data'; import { toneText } from '@/components/Data'; import type { ChatMessage } from '../types'; @@ -13,6 +13,8 @@ import { markUserBackgroundJob } from '../../../background-job-origin'; type BackgroundTaskTrayProps = { messages: ChatMessage[]; + /** Which conversation the chips belong to. A restart emits its events back into this session. */ + sessionId: string | null; }; /** Same vocabulary as the transcript's own task rows — a completed task must not be green here and grey there. */ @@ -31,7 +33,7 @@ function statusTone(status: BackgroundTask['status']): Tone { * Clicking a chip opens what it is actually doing right now: a subagent's own tool-by-tool trace, or the * tail of a backgrounded shell's log, both read from the file Claude Code streams the task into. */ -export const BackgroundTaskTray = ({ messages }: BackgroundTaskTrayProps) => { +export const BackgroundTaskTray = ({ messages, sessionId }: BackgroundTaskTrayProps) => { const { tasks, pinned, unpinned, running, clearable, togglePin, dismiss, dismissFinished } = useBackgroundTasks(messages); const [openId, setOpenId] = useState(null); @@ -56,6 +58,46 @@ export const BackgroundTaskTray = ({ messages }: BackgroundTaskTrayProps) => { } }; + /** + * Run this job's command again, without asking the agent. + * + * The platform recovers the command from the transcript, stops whatever is running, and respawns it + * appending to the same output file — so the log keeps growing rather than starting a new one, and Stop + * keeps working because it finds the process by who holds that file open. + * + * Offered for a FINISHED job too, which is the case that matters most: a dev server that fell over is + * exactly the thing you want back without typing a sentence at an agent that may be mid-turn. + */ + const restartTask = async (taskId: string) => { + try { + const result = await client.post<{ ok: boolean; reason?: string }>(`/chat/tasks/${taskId}/restart`, { + sessionId, + }); + if (!result.ok) toast.error(result.reason ?? 'Could not restart that job'); + else markUserBackgroundJob(taskId); + } catch (err) { + toast.error(err instanceof Error ? err.message : 'Could not restart that job'); + } + }; + + /** + * Freeze or thaw the process, without ending it. + * + * SIGSTOP/SIGCONT, so a dev server keeps its port, its memory and its place — a paused one does not + * refuse connections, it stops answering them, which is the honest thing for the button to do and worth + * knowing before you wonder why the page hangs. + */ + const pauseTask = async (taskId: string, paused: boolean): Promise => { + try { + const result = await client.post<{ ok: boolean; reason?: string }>(`/chat/tasks/${taskId}/pause`, { paused }); + if (!result.ok) toast.error(result.reason ?? 'Could not pause that job'); + return result.ok; + } catch (err) { + toast.error(err instanceof Error ? err.message : 'Could not pause that job'); + return false; + } + }; + // Derived, not stored: dismissing the chip that is open closes the panel without a second piece of state // having to be kept in step with the first. const open = tasks.find((t) => t.taskId === openId) ?? null; @@ -80,6 +122,8 @@ export const BackgroundTaskTray = ({ messages }: BackgroundTaskTrayProps) => { task={open} onClose={() => setOpenId(null)} onStop={open.status ? undefined : () => void stopTask(open.taskId)} + onPause={(next) => pauseTask(open.taskId, next)} + onRestart={() => void restartTask(open.taskId)} /> )} @@ -234,12 +278,20 @@ const TaskChip = ({ task, active, pinned, onClick, onTogglePin, onDismiss, onSto type TaskPanelProps = { /** Present only while the task is running — the panel shows a real button rather than a bare glyph. */ onStop?: () => void; + /** Freeze and thaw the process. Present while it is running; the label follows the state. */ + onPause?: (paused: boolean) => Promise; + /** Run the command again, whether it is running or long finished. */ + onRestart?: () => void; task: BackgroundTask; onClose: () => void; }; -const TaskPanel = ({ task, onClose, onStop }: TaskPanelProps) => { +const TaskPanel = ({ task, onClose, onStop, onPause, onRestart }: TaskPanelProps) => { const running = !task.status; + // Local, because a paused process is a short-lived state somebody is watching. It resets on reload, + // which is honest enough: SIGCONT is idempotent, and the button says what it will do rather than what + // it believes has happened. + const [paused, setPaused] = useState(false); const { data, isLoading } = useTaskDetail(task.taskId, running); const bodyRef = useRef(null); const pinnedRef = useRef(true); @@ -264,6 +316,33 @@ const TaskPanel = ({ task, onClose, onStop }: TaskPanelProps) => { {/* Labelled here, unlike the chip's icon-only version. The panel is where you come to decide whether to keep something running, so the control that ends it should not be a glyph you have to hover to identify. */} + {/* Pause and Restart sit beside Stop rather than on the chip: the panel is where you decide what + to do with a job, and a row of unlabelled glyphs on a pill is where you misclick. */} + {running && onPause && ( + + )} + {onRestart && ( + + )} {running && onStop && (