pause, resume and restart a background job without asking the agent
Three buttons in the expanded task panel, beside Stop. Not 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. ── Pause / Resume ── SIGSTOP and SIGCONT — the only pair that means "stop doing things and keep everything". The process keeps its memory, its open files and its bound port, and picks up mid-instruction. Signals go to every pid holding the job's output file, because for a shell job that is the shell AND what it spawned, and freezing one of them would be worse than freezing neither. Verified by watching a job that writes a line a second: 3 lines, frozen at 3 across three seconds of pause, 6 after resume. Worth knowing before pressing it: a paused server does not REFUSE connections, it stops answering them, so a browser pointed at it hangs rather than erroring. That is what pausing is, not a shortcoming here. ── Restart, and the cost it accepts ── The command is recovered by joining the transcript to itself: the harness records the `tool_use` carrying `command`, and separately its result — the text `Command running in background with ID: <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 searched for, since a task file at `<tmp>/claude-<uid>/<slug>/<session>/tasks/<id>.output` names both. The respawned process appends to that same output file. 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. The real cost, chosen deliberately: the platform becomes a SECOND owner of a task. The harness stops watching when its own child dies, so the platform supervises what it spawned and emits the same `task:started` / `task:notification` pair everything downstream is built on. `emitToSession` is the seam, and it is the only one — nothing else learns that a task can have two kinds of parent. A restart re-announces an id already in the transcript, so `task:started` now RESETS an existing task row rather than appending a second one. Without that the tray showed the job twice, once finished and once running. Restart is offered for a finished job too, which is the case that matters most: a dev server that fell over is exactly what you want back without typing a sentence at an agent that may be mid-turn. Verified end to end against a planted task: command recovered, old process killed, new one spawned, output APPENDED (`first run output` then `RESTARTED-OK`), exit reported through the callback, supervision registry released afterwards. `[open]` A restarted job does not survive an `officer` restart — the platform is its parent now. The harness's own jobs never did either. tsgo clean, frontend builds, 890 pass / 7 fail unchanged. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -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'));
|
||||
|
||||
@@ -711,7 +711,7 @@ export type BackgroundTaskDetail =
|
||||
* `<projects>/<slug>/<session>/subagents/agent-<id>.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/<pid>/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<number[]> {
|
||||
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) {
|
||||
|
||||
@@ -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 `<task-id>.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<string, { proc: Bun.Subprocess }>();
|
||||
|
||||
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: <task 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
|
||||
* `<tmp>/claude-<uid>/<project-slug>/<session-uuid>/tasks/<id>.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<string | null> {
|
||||
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<string, string>();
|
||||
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<string, unknown>[]) {
|
||||
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<RestartResult> {
|
||||
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 };
|
||||
}
|
||||
@@ -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<WSData>, msg);
|
||||
return true;
|
||||
}
|
||||
|
||||
async function handleChat(
|
||||
ws: ServerWebSocket<WSData>,
|
||||
msg: {
|
||||
|
||||
Reference in New Issue
Block a user