feat: post-review backlog hardening (cancel/parser/stall/history/9502)

Five independent items from the post-review backlog. F1: Stop on an external
agent task now aborts the running child via a per-task AbortController registry
reachable from the cancel route, and finalizes the assistant message as
cancelled (fixing two latent bugs — catch blocks left the message streaming,
and warm success-paths wrote complete on an aborted turn); warm pools/worktrees
are preserved and the native path is unchanged. F2/F3: prune the tool-call
parser to its two load-bearing exports (unexport eight zero-caller symbols, add
a gate test for the <invoke>-as-text fallback) and route placeholder-rejection
logging through pino. F6: a 90s per-chunk stall-timeout wraps native inference's
fullStream via AbortSignal.any so a hung stream finalizes the message instead of
hanging — no retry (a pure classifyStreamError helper is added). F7: a read-only
view_session_history MCP tool (newest-N, chronological). F9: retire the unused
apps/coder/web :9502 fallback SPA, keeping every API/WS/health/MCP route.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-03 02:23:11 +00:00
parent 9a139633b8
commit f32fd928b3
48 changed files with 1014 additions and 2254 deletions

View File

@@ -1,12 +1,5 @@
import { resolve, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';
import { existsSync } from 'node:fs';
import Fastify from 'fastify';
import fastifyWebsocket from '@fastify/websocket';
import fastifyStatic from '@fastify/static';
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
import { loadConfig } from './config.js';
import { getSql, applySchema, pingDb, closeDb } from './db.js';
import { startMcpServer } from './services/mcp-server.js';
@@ -257,7 +250,7 @@ async function main() {
registerPendingRoutes(app, sql);
registerCheckpointRoutes(app, sql);
registerAgentSessionRoutes(app, sql);
registerTaskRoutes(app, sql, inferenceApi);
registerTaskRoutes(app, sql, inferenceApi, dispatcher.cancelExternalTask);
registerInboxRoutes(app, sql);
registerStatsRoutes(app, sql);
registerArenaRoutes(app, sql);
@@ -266,28 +259,6 @@ async function main() {
registerLifecycleRoutes(app, sql);
registerWebSocket(app, sql, broker);
// Serve static frontend (built web app). In production, the dist/ is
// copied to ../web relative to the dist/ directory at /app/web. In dev,
// check adjacent to the source.
const webRoot = resolve(__dirname, '../web');
if (existsSync(webRoot)) {
await app.register(fastifyStatic, {
root: webRoot,
prefix: '/',
// Don't intercept /api routes — static only serves files that exist.
wildcard: false,
});
// SPA fallback: serve index.html for non-API routes that don't match a file.
app.setNotFoundHandler(async (req, reply) => {
if (req.url.startsWith('/api')) {
reply.code(404);
return { error: 'not found' };
}
return reply.sendFile('index.html');
});
app.log.info(`serving frontend from ${webRoot}`);
}
// Graceful shutdown
const shutdown = async () => {
app.log.info('shutting down');

View File

@@ -0,0 +1,138 @@
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { readFileSync } from 'node:fs';
import { resolve } from 'node:path';
import Fastify, { type FastifyInstance } from 'fastify';
import postgres from 'postgres';
import { registerTaskRoutes } from '../tasks.js';
/**
* F1 — POST /api/tasks/:id/cancel route wiring.
*
* The route's job: reach the in-flight external run via `cancelExternal(taskId)`
* (the new abort hook), keep cancelling native inference for open chats unchanged,
* and land the task row in 'cancelled'. The streaming assistant message is
* finalized by the dispatcher's run-function, not here — that path is covered by
* finalize-message.test.ts. This suite pins the route's behavior against a real DB.
*/
describe.runIf(!!process.env.DATABASE_URL)('POST /api/tasks/:id/cancel (route, F1)', () => {
let sql: ReturnType<typeof postgres>;
let app: FastifyInstance;
let projectId: string;
let sessionId: string;
let chatId: string;
const externalCancelCalls: string[] = [];
const inferenceCancelCalls: Array<[string, string]> = [];
let externalReturns = true;
beforeAll(async () => {
sql = postgres(process.env.DATABASE_URL!, { max: 3 });
const serverSchema = resolve(__dirname, '../../../../server/src/schema.sql');
const coderSchema = resolve(__dirname, '../../schema.sql');
await sql.unsafe(readFileSync(serverSchema, 'utf8'));
await sql.unsafe(readFileSync(coderSchema, 'utf8'));
const [p] = await sql<{ id: string }[]>`
INSERT INTO projects (name, path, status) VALUES ('f1-cancel-route', '/tmp/f1-cancel-route', 'open') RETURNING id
`;
projectId = p!.id;
const [s] = await sql<{ id: string }[]>`
INSERT INTO sessions (project_id, name, model, status) VALUES (${projectId}, 'f1', 'm', 'open') RETURNING id
`;
sessionId = s!.id;
const [c] = await sql<{ id: string }[]>`
INSERT INTO chats (session_id, name, status) VALUES (${sessionId}, 'tab', 'open') RETURNING id
`;
chatId = c!.id;
app = Fastify();
registerTaskRoutes(
app,
sql,
{
cancel: async (sid: string, cid: string) => {
inferenceCancelCalls.push([sid, cid]);
return false;
},
},
(taskId: string) => {
externalCancelCalls.push(taskId);
return externalReturns;
},
);
await app.ready();
});
afterAll(async () => {
if (app) await app.close();
if (!sql) return;
await sql`DELETE FROM messages WHERE session_id = ${sessionId}`.catch(() => {});
await sql`DELETE FROM tasks WHERE project_id = ${projectId}`.catch(() => {});
await sql`DELETE FROM chats WHERE id = ${chatId}`.catch(() => {});
await sql`DELETE FROM sessions WHERE id = ${sessionId}`.catch(() => {});
await sql`DELETE FROM projects WHERE id = ${projectId}`.catch(() => {});
await sql.end({ timeout: 5 });
});
async function insertTask(agent: string | null, state: string): Promise<string> {
const [t] = await sql<{ id: string }[]>`
INSERT INTO tasks (project_id, input, agent, session_id, state, started_at)
VALUES (${projectId}, 'do a thing', ${agent}, ${sessionId}, ${state}, clock_timestamp())
RETURNING id
`;
return t!.id;
}
it('reaches cancelExternal and lands the task cancelled for a running external task', async () => {
externalReturns = true;
externalCancelCalls.length = 0;
const taskId = await insertTask('opencode', 'running');
const res = await app.inject({ method: 'POST', url: `/api/tasks/${taskId}/cancel` });
expect(res.statusCode).toBe(200);
expect(res.json()).toEqual({ cancelled: true });
expect(externalCancelCalls).toContain(taskId);
const [row] = await sql<{ state: string; ended_at: Date | null }[]>`
SELECT state, ended_at FROM tasks WHERE id = ${taskId}
`;
expect(row!.state).toBe('cancelled');
expect(row!.ended_at).not.toBeNull();
});
it('still cancels a native boocode task (cancelExternal returns false → inference.cancel path unchanged)', async () => {
externalReturns = false; // native task: no controller registered
externalCancelCalls.length = 0;
inferenceCancelCalls.length = 0;
const taskId = await insertTask(null, 'running');
const res = await app.inject({ method: 'POST', url: `/api/tasks/${taskId}/cancel` });
expect(res.statusCode).toBe(200);
// The route calls cancelExternal unconditionally (cheap, returns false here)...
expect(externalCancelCalls).toContain(taskId);
// ...and the native inference.cancel path still fires for the open chat.
expect(inferenceCancelCalls).toContainEqual([sessionId, chatId]);
const [row] = await sql<{ state: string }[]>`SELECT state FROM tasks WHERE id = ${taskId}`;
expect(row!.state).toBe('cancelled');
});
it('rejects cancelling an already-terminal task with 409 and never touches the abort hook', async () => {
externalCancelCalls.length = 0;
const taskId = await insertTask('opencode', 'completed');
const res = await app.inject({ method: 'POST', url: `/api/tasks/${taskId}/cancel` });
expect(res.statusCode).toBe(409);
expect(externalCancelCalls).not.toContain(taskId);
});
it('returns 404 for an unknown task', async () => {
const res = await app.inject({
method: 'POST',
url: `/api/tasks/00000000-0000-0000-0000-000000000000/cancel`,
});
expect(res.statusCode).toBe(404);
});
});

View File

@@ -8,6 +8,12 @@ interface InferenceApi {
cancel: (sessionId: string, chatId: string) => Promise<boolean>;
}
// F1: the dispatcher's reach into an in-flight external-agent run. Narrow by
// design (not the whole dispatcher) — the route only needs to fire the abort.
// Returns true when a controller was registered for the task (an external run was
// in flight), false otherwise (native boocode task, or already finished).
export type ExternalCancelFn = (taskId: string) => boolean;
const CreateBody = z.object({
project_id: z.string().uuid(),
input: z.string().min(1).max(64_000),
@@ -27,7 +33,12 @@ const ListQuery = z.object({
project_id: z.string().uuid().optional(),
});
export function registerTaskRoutes(app: FastifyInstance, sql: Sql, inference: InferenceApi): void {
export function registerTaskRoutes(
app: FastifyInstance,
sql: Sql,
inference: InferenceApi,
cancelExternal: ExternalCancelFn,
): void {
// POST /api/tasks — create a new task
app.post('/api/tasks', async (req, reply) => {
const parsed = CreateBody.safeParse(req.body);
@@ -127,7 +138,14 @@ export function registerTaskRoutes(app: FastifyInstance, sql: Sql, inference: In
cancelPendingPermission(taskId);
// If running, try to cancel inference
// F1: abort the in-flight external-agent run (opencode / goose / qwen / claude).
// Idempotent — a double-Stop re-aborts harmlessly; a native boocode task is not
// registered, so this returns false and the inference.cancel path below handles
// it unchanged. The dispatcher's run-function finalizes the streaming assistant
// message as 'cancelled' once the backend honors the signal.
cancelExternal(taskId);
// If running, try to cancel inference (native boocode path — unchanged).
if ((task.state === 'running' || task.state === 'blocked') && task.session_id) {
// Find active chat in the task's session
const chats = await sql<{ id: string }[]>`

View File

@@ -0,0 +1,51 @@
import { describe, it, expect } from 'vitest';
import { createCancelRegistry } from '../cancel-registry.js';
/**
* F1 — per-task abort wiring. The registry is the missing link between the Stop
* route and the in-flight external run: register an AbortController per task id,
* cancel(taskId) aborts its signal, the run's .finally deletes it. Pure (no DB /
* child / IO) so the abort + idempotency contract is unit-testable in isolation.
*/
describe('CancelRegistry (F1 abort wiring)', () => {
it('register hands back a fresh controller; cancel aborts its signal', () => {
const reg = createCancelRegistry();
const ac = reg.register('t1');
expect(ac.signal.aborted).toBe(false);
expect(reg.has('t1')).toBe(true);
expect(reg.cancel('t1')).toBe(true);
expect(ac.signal.aborted).toBe(true);
});
it('cancel on an unknown task returns false (native task / cancel-before-register)', () => {
const reg = createCancelRegistry();
expect(reg.has('nope')).toBe(false);
expect(reg.cancel('nope')).toBe(false);
});
it('double-Stop is idempotent: a second cancel never throws and the signal stays aborted', () => {
const reg = createCancelRegistry();
const ac = reg.register('t1');
expect(reg.cancel('t1')).toBe(true);
// The run-function has not hit its .finally yet, so the entry is still
// present — a rapid second Stop re-aborts (abort() no-ops) without throwing.
expect(() => reg.cancel('t1')).not.toThrow();
expect(reg.cancel('t1')).toBe(true);
expect(ac.signal.aborted).toBe(true);
});
it('cancel after delete returns false (cancel-after-natural-exit is safe)', () => {
const reg = createCancelRegistry();
reg.register('t1');
reg.delete('t1');
expect(reg.has('t1')).toBe(false);
expect(reg.cancel('t1')).toBe(false);
});
it('delete of an unknown id is a no-op (never throws)', () => {
const reg = createCancelRegistry();
expect(() => reg.delete('ghost')).not.toThrow();
});
});

View File

@@ -0,0 +1,163 @@
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { readFileSync } from 'node:fs';
import { resolve } from 'node:path';
import postgres from 'postgres';
import type { WsFrame } from '@boocode/contracts/ws-frames';
import { classifyTerminalStatus, finalizeStreamingMessage } from '../finalize-message.js';
/**
* F1 (D-7 / OCE-001 / OCE-002) — finalizing a Stop'd or errored external turn.
*
* `classifyTerminalStatus` is the pure D-7 decision (user Stop / AbortError →
* cancelled, genuine error → failed). `finalizeStreamingMessage` writes that
* terminal state onto the streaming assistant row and publishes the matching
* message_complete frame — idempotently, guarded by `WHERE status='streaming'`,
* so a double-Stop or an abort-then-catch settles the message exactly once and
* never clobbers a row that already finished cleanly.
*/
describe('classifyTerminalStatus (pure, D-7)', () => {
it('maps a fired abort signal to cancelled (user Stop)', () => {
expect(classifyTerminalStatus({ aborted: true })).toBe('cancelled');
});
it('maps a thrown AbortError to cancelled', () => {
const e = new Error('the operation was aborted');
e.name = 'AbortError';
expect(classifyTerminalStatus({ aborted: false, error: e })).toBe('cancelled');
});
it('maps a genuine thrown error to failed', () => {
expect(classifyTerminalStatus({ aborted: false, error: new Error('boom') })).toBe('failed');
});
it('defaults a no-abort / no-error catch to failed', () => {
expect(classifyTerminalStatus({ aborted: false })).toBe('failed');
});
});
describe.runIf(!!process.env.DATABASE_URL)('finalizeStreamingMessage (DB)', () => {
let sql: ReturnType<typeof postgres>;
let projectId: string;
let sessionId: string;
let chatId: string;
beforeAll(async () => {
sql = postgres(process.env.DATABASE_URL!, { max: 3 });
// Server schema owns messages/sessions/chats (FK targets); coder schema after.
const serverSchema = resolve(__dirname, '../../../../server/src/schema.sql');
const coderSchema = resolve(__dirname, '../../schema.sql');
await sql.unsafe(readFileSync(serverSchema, 'utf8'));
await sql.unsafe(readFileSync(coderSchema, 'utf8'));
const [p] = await sql<{ id: string }[]>`
INSERT INTO projects (name, path, status) VALUES ('f1-finalize', '/tmp/f1-finalize', 'open') RETURNING id
`;
projectId = p!.id;
const [s] = await sql<{ id: string }[]>`
INSERT INTO sessions (project_id, name, model, status) VALUES (${projectId}, 'f1', 'm', 'open') RETURNING id
`;
sessionId = s!.id;
const [c] = await sql<{ id: string }[]>`
INSERT INTO chats (session_id, name, status) VALUES (${sessionId}, 'tab', 'open') RETURNING id
`;
chatId = c!.id;
});
afterAll(async () => {
if (!sql) return;
await sql`DELETE FROM messages WHERE session_id = ${sessionId}`.catch(() => {});
await sql`DELETE FROM chats WHERE id = ${chatId}`.catch(() => {});
await sql`DELETE FROM sessions WHERE id = ${sessionId}`.catch(() => {});
await sql`DELETE FROM projects WHERE id = ${projectId}`.catch(() => {});
await sql.end({ timeout: 5 });
});
async function insertStreaming(): Promise<string> {
const [m] = await sql<{ id: string }[]>`
INSERT INTO messages (session_id, chat_id, role, content, status)
VALUES (${sessionId}, ${chatId}, 'assistant', '', 'streaming') RETURNING id
`;
return m!.id;
}
it('finalizes a streaming row to cancelled, persists partial content, publishes one frame', async () => {
const id = await insertStreaming();
const frames: WsFrame[] = [];
const did = await finalizeStreamingMessage(sql, (_s, f) => frames.push(f), {
sessionId,
chatId,
assistantId: id,
status: 'cancelled',
model: 'qwen',
content: 'partial answer',
});
expect(did).toBe(true);
const [row] = await sql<{ status: string; content: string; finished_at: Date | null }[]>`
SELECT status, content, finished_at FROM messages WHERE id = ${id}
`;
expect(row!.status).toBe('cancelled');
expect(row!.content).toBe('partial answer');
expect(row!.finished_at).not.toBeNull();
expect(frames).toHaveLength(1);
expect(frames[0]!.type).toBe('message_complete');
expect((frames[0] as { status?: string }).status).toBe('cancelled');
});
it('is idempotent for a double-Stop: second call updates nothing and re-publishes nothing', async () => {
const id = await insertStreaming();
const frames: WsFrame[] = [];
const push = (_s: string, f: WsFrame): void => {
frames.push(f);
};
expect(
await finalizeStreamingMessage(sql, push, { sessionId, chatId, assistantId: id, status: 'cancelled', model: null }),
).toBe(true);
expect(
await finalizeStreamingMessage(sql, push, { sessionId, chatId, assistantId: id, status: 'cancelled', model: null }),
).toBe(false);
expect(frames).toHaveLength(1);
const [row] = await sql<{ status: string }[]>`SELECT status FROM messages WHERE id = ${id}`;
expect(row!.status).toBe('cancelled');
});
it('never clobbers a row that already finished cleanly (abort raced a clean finish)', async () => {
const [m] = await sql<{ id: string }[]>`
INSERT INTO messages (session_id, chat_id, role, content, status)
VALUES (${sessionId}, ${chatId}, 'assistant', 'done', 'complete') RETURNING id
`;
const id = m!.id;
const frames: WsFrame[] = [];
const did = await finalizeStreamingMessage(sql, (_s, f) => frames.push(f), {
sessionId,
chatId,
assistantId: id,
status: 'cancelled',
model: null,
});
expect(did).toBe(false);
expect(frames).toHaveLength(0);
const [row] = await sql<{ status: string; content: string }[]>`
SELECT status, content FROM messages WHERE id = ${id}
`;
expect(row!.status).toBe('complete');
expect(row!.content).toBe('done');
});
it('no-ops on an empty assistantId (throw happened before the row was created)', async () => {
const frames: WsFrame[] = [];
const did = await finalizeStreamingMessage(sql, (_s, f) => frames.push(f), {
sessionId,
chatId,
assistantId: '',
status: 'failed',
model: null,
});
expect(did).toBe(false);
expect(frames).toHaveLength(0);
});
});

View File

@@ -0,0 +1,50 @@
/**
* F1 — per-task abort registry. A Stop on an external-agent task must reach the
* in-flight run and abort its child / prompt. Each external run-function registers
* its per-turn AbortController here keyed by task id; the cancel route calls
* `cancel(taskId)` to fire it; the run-function's `.finally` deletes the entry.
*
* Idempotent by construction:
* - `cancel()` on an already-aborted controller no-ops (AbortController.abort()
* is idempotent) → a rapid double-Stop is safe.
* - `cancel()` on an unknown / already-finished task returns false → a
* cancel-after-natural-exit (entry already deleted) and a Stop on a native
* boocode task (never registered) are both safe no-ops.
*
* Pure (no DB / child / IO) so the abort wiring + idempotency contract is
* unit-testable in isolation — mirrors the turn-guard / lifecycle-decisions
* pure-helper precedent.
*/
export interface CancelRegistry {
/** Create + store an AbortController for this task, returning it for the run. */
register(taskId: string): AbortController;
/** Abort the task's in-flight run. Returns false when no controller is registered. */
cancel(taskId: string): boolean;
/** Drop the task's entry (called from the run's `.finally`). No-op if absent. */
delete(taskId: string): void;
/** Whether a controller is currently registered for this task. */
has(taskId: string): boolean;
}
export function createCancelRegistry(): CancelRegistry {
const controllers = new Map<string, AbortController>();
return {
register(taskId) {
const ac = new AbortController();
controllers.set(taskId, ac);
return ac;
},
cancel(taskId) {
const ac = controllers.get(taskId);
if (!ac) return false;
ac.abort();
return true;
},
delete(taskId) {
controllers.delete(taskId);
},
has(taskId) {
return controllers.has(taskId);
},
};
}

View File

@@ -22,6 +22,12 @@ import { shouldUseClaudeSdk } from './backends/claude-sdk-routing.js';
import type { AgentBackend, AgentEvent } from './agent-backend.js';
import { publishAgentStatus } from './agent-status-publish.js';
import type { AgentStatus } from './normalize-agent-status.js';
import { createCancelRegistry } from './cancel-registry.js';
import {
finalizeStreamingMessage,
classifyTerminalStatus,
type TerminalMessageStatus,
} from './finalize-message.js';
interface InferenceRunner {
enqueue: (sessionId: string, chatId: string, assistantId: string, user: string) => void;
@@ -43,7 +49,11 @@ interface Deps {
const POLL_INTERVAL_MS = 2_000;
const COMPLETION_POLL_MS = 2_000;
export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<void> } {
export function createDispatcher(deps: Deps): {
cancelExternalTask(taskId: string): boolean;
start(): void;
stop(): Promise<void>;
} {
const { sql, inference, broker, log, config } = deps;
let timer: ReturnType<typeof setInterval> | null = null;
let listener: { unlisten: () => Promise<void> } | null = null;
@@ -55,6 +65,13 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
// turn at a time.
const inflight = new Map<string, Promise<void>>();
// F1: per-task abort registry. Each external run-function registers its per-turn
// AbortController here (keyed by task id); the cancel route reaches it through the
// exported `cancelExternalTask`; the run's `.finally` deletes the entry. Native
// boocode tasks are never registered, so a Stop on one returns false and falls
// through to the unchanged inference.cancel path.
const taskControllers = createCancelRegistry();
// Shared entry point for both the poll timer and the NOTIFY listener. poll()'s
// `polling`/`stopping` guard makes this safe to call concurrently — a notify
// arriving mid-poll returns immediately and never double-dispatches.
@@ -83,6 +100,40 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
publishAgentStatus(broker.publishFrame, sessionId, chatId, agent, status, reason);
}
// F1 (OCE-001/OCE-002): finalize a streaming assistant message into a terminal
// state and publish the matching message_complete frame. Best-effort + idempotent
// (the helper's `WHERE status='streaming'` guard) — a failure here must never mask
// the original abort/error, so it logs and swallows.
function finalizeMessage(
sessionId: string,
chatId: string,
assistantId: string,
status: TerminalMessageStatus,
model: string | null,
content?: string,
): Promise<boolean> {
return finalizeStreamingMessage(sql, broker.publishFrame, {
sessionId,
chatId,
assistantId,
status,
model,
content,
}).catch((err) => {
log.error({ err: err instanceof Error ? err.message : String(err), assistantId }, 'dispatcher: finalizeStreamingMessage failed');
return false;
});
}
// F1: the cancel route's reach into an in-flight external run. Idempotent — a
// double-Stop re-aborts an already-aborted controller (no-op) and a Stop on a
// finished/native task returns false. Aborting only fires the backend's per-turn
// cancel (session.abort / session/cancel / interrupt / child.kill); it never kills
// a warm pool process, so persistent worktrees + pooled backends are preserved.
function cancelExternalTask(taskId: string): boolean {
return taskControllers.cancel(taskId);
}
async function poll(): Promise<void> {
// `polling` serializes poll() execution itself (timer + NOTIFY can fire
// concurrently) so we never double-select a task. It does NOT serialize task
@@ -116,6 +167,9 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
// with the same key is skipped and a concurrent poll can't re-pick it.
const p = runTask(task).finally(() => {
inflight.delete(key);
// F1: drop the abort controller once the run settles. After this, a Stop
// on the (now-finished) task returns false — cancel-after-exit is safe.
taskControllers.delete(task.id);
});
inflight.set(key, p);
}
@@ -312,13 +366,16 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
return;
}
// Create an abort controller for this task
const ac = new AbortController();
// F1: register the per-task abort controller so a Stop reaches this run.
const ac = taskControllers.register(taskId);
// #10: hoisted above the try so the catch block can report `error` status with
// the (chat, agent) key. Empty until resolved below; guarded before use.
let sessionId = '';
let chatId = '';
// F1: hoisted so the catch / abort short-circuit can finalize the streaming
// assistant row. Empty until the row is created; finalize no-ops on ''.
let assistantId = '';
try {
// Mark running
@@ -384,7 +441,7 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
VALUES (${sessionId}, ${chatId}, 'assistant', '', 'streaming', ${task.model}, clock_timestamp())
RETURNING id
`;
const assistantId = assistantMsg!.id;
assistantId = assistantMsg!.id;
// write-edit-robustness #4: pre-turn worktree checkpoint (best-effort; a
// failure logs and never breaks dispatch). This path uses a per-task worktree
@@ -526,6 +583,20 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
}
}
// F1: abort short-circuit BEFORE the unconditional 'complete' write. A Stop
// (cancelExternalTask → ac.abort) or shutdown finalizes the streaming row as
// 'cancelled' (keeping whatever streamed) instead of recording 'complete',
// and skips the diff. This one-shot path owns a per-task worktree, so we DO
// tear it down here (unlike the warm paths, which keep their persistent one).
if (ac.signal.aborted || stopping) {
await finalizeMessage(sessionId, chatId, assistantId, 'cancelled', task.model, assistantContent);
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
emitAgentStatus(sessionId, chatId, agent, 'idle', stopping ? 'shutdown' : 'cancelled');
await cleanupWorktree(projectPath, taskId);
clearTaskCommands(taskId);
return;
}
await sql`
UPDATE messages
SET content = ${assistantContent}, status = 'complete', finished_at = clock_timestamp()
@@ -539,14 +610,6 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
model: task.model,
} as WsFrame);
if (stopping) {
await sql`
UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}
`;
await cleanupWorktree(projectPath, taskId);
return;
}
// Step 3: Diff the worktree and queue pending changes
log.info({ taskId }, 'dispatcher: diffing worktree');
const diff = await diffWorktree(worktreePath, projectPath, { signal: ac.signal });
@@ -587,18 +650,26 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
const status = classifyTerminalStatus({ aborted: ac.signal.aborted, error: err });
log.error({ taskId, agent, err: errMsg }, 'dispatcher: external agent error');
// Guard `NOT IN ('cancelled','completed')` so a genuine error in the catch
// never overwrites a state the cancel route already wrote (user-Stop wins).
await sql`
UPDATE tasks
SET state = 'failed', ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId}
SET state = ${status}, ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId} AND state NOT IN ('cancelled', 'completed')
`.catch(() => {});
// F1 (OCE-001): finalize the streaming assistant message — the catch
// previously updated only `tasks` and left the message 'streaming' forever
// (the BooChat 5-min sweep runs in a different process and can't reach it).
await finalizeMessage(sessionId, chatId, assistantId, status, task.model);
// #10: external-agent turn failed/crashed. chatId may be unbound if the throw
// preceded its assignment — guard so the status publish never masks the real
// error.
if (chatId) emitAgentStatus(sessionId, chatId, agent, 'error', 'failed');
if (chatId) emitAgentStatus(sessionId, chatId, agent, status === 'cancelled' ? 'idle' : 'error', status === 'cancelled' ? 'cancelled' : 'failed');
// Best-effort cleanup
await cleanupWorktree(projectPath, taskId);
@@ -652,11 +723,14 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
return;
}
const ac = new AbortController();
// F1: register the per-task abort controller so a Stop reaches this run.
const ac = taskControllers.register(taskId);
// #10: hoisted so the catch can report `error` with the (chat, agent) key.
let sessionId = '';
let chatId = '';
// F1: hoisted so the catch / abort short-circuit can finalize the streaming row.
let assistantId = '';
try {
// execution_path = 'acp' — the schema CHECK has no 'opencode_server' value
@@ -728,7 +802,7 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
VALUES (${sessionId}, ${chatId}, 'assistant', '', 'streaming', ${task.model}, clock_timestamp())
RETURNING id
`;
const assistantId = assistantMsg!.id;
assistantId = assistantMsg!.id;
// write-edit-robustness #4: pre-turn checkpoint of the persistent session
// worktree (best-effort; never breaks dispatch). worktreeId comes from the
@@ -856,6 +930,18 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
await persistExternalAgentTurn(sql, assistantId, [...toolSnaps.values()], reasoningText);
// F1: abort short-circuit BEFORE the unconditional 'complete' write — fixes
// the warm success-path recording 'complete' on a Stop'd turn. The abort fired
// session.abort on the prompt only: the persistent session worktree is kept
// (no cleanup) and the pooled opencode server stays warm for the next turn.
if (ac.signal.aborted || stopping) {
await finalizeMessage(sessionId, chatId, assistantId, 'cancelled', task.model, assistantContent);
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
emitAgentStatus(sessionId, chatId, agent, 'idle', stopping ? 'shutdown' : 'cancelled');
clearTaskCommands(taskId);
return; // worktree persists (no cleanup); backend stays warm
}
await sql`
UPDATE messages
SET content = ${assistantContent}, status = 'complete', finished_at = clock_timestamp()
@@ -868,11 +954,6 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
model: task.model,
} as WsFrame);
if (stopping) {
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
return; // worktree persists (no cleanup); backend stays warm
}
// 1.10: diff the persistent worktree against its captured baseline and
// SUPERSEDE the session's prior pending row (latest-wins, one accumulating
// diff) instead of stacking. Stamp agent for DiffPanel attribution.
@@ -920,14 +1001,17 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
clearTaskCommands(taskId);
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
const status = classifyTerminalStatus({ aborted: ac.signal.aborted, error: err });
log.error({ taskId, agent, err: errMsg }, 'dispatcher: opencode server error');
await sql`
UPDATE tasks
SET state = 'failed', ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId}
SET state = ${status}, ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId} AND state NOT IN ('cancelled', 'completed')
`.catch(() => {});
// F1 (OCE-001): finalize the streaming message (was left 'streaming').
await finalizeMessage(sessionId, chatId, assistantId, status, task.model);
// #10: turn crashed.
if (chatId) emitAgentStatus(sessionId, chatId, agent, 'error', 'crashed');
if (chatId) emitAgentStatus(sessionId, chatId, agent, status === 'cancelled' ? 'idle' : 'error', status === 'cancelled' ? 'cancelled' : 'crashed');
clearTaskCommands(taskId);
// No worktree cleanup (persistent); backend stays warm for the next turn.
}
@@ -988,7 +1072,10 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
return;
}
const ac = new AbortController();
// F1: register the per-task abort controller so a Stop reaches this run.
const ac = taskControllers.register(taskId);
// F1: hoisted so the catch / abort short-circuit can finalize the streaming row.
let assistantId = '';
try {
await sql`
@@ -1010,7 +1097,7 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
VALUES (${sessionId}, ${chatId}, 'assistant', '', 'streaming', ${task.model}, clock_timestamp())
RETURNING id
`;
const assistantId = assistantMsg!.id;
assistantId = assistantMsg!.id;
// write-edit-robustness #4: pre-turn checkpoint of the persistent session
// worktree (best-effort; never breaks dispatch). Same worktree the opencode
@@ -1121,6 +1208,18 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
await persistExternalAgentTurn(sql, assistantId, [...toolSnaps.values()], reasoningText);
// F1: abort short-circuit BEFORE the unconditional 'complete' write — fixes
// the warm success-path recording 'complete' on a Stop'd turn. The abort fired
// session/cancel on the warm connection only (never killed the child), so the
// persistent worktree is kept and the pooled (chat,agent) backend stays warm.
if (ac.signal.aborted || stopping) {
await finalizeMessage(sessionId, chatId, assistantId, 'cancelled', task.model, assistantContent);
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
emitAgentStatus(sessionId, chatId, agent, 'idle', stopping ? 'shutdown' : 'cancelled');
clearTaskCommands(taskId);
return; // worktree persists (no cleanup); backend stays warm
}
await sql`
UPDATE messages
SET content = ${assistantContent}, status = 'complete', finished_at = clock_timestamp()
@@ -1133,11 +1232,6 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
model: task.model,
} as WsFrame);
if (stopping) {
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
return; // worktree persists (no cleanup); backend stays warm
}
// Diff the persistent worktree against its captured baseline and SUPERSEDE
// the session's prior pending row (latest-wins) — identical to opencode.
const diff = await diffWorktree(worktreePath, projectPath, {
@@ -1184,14 +1278,17 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
clearTaskCommands(taskId);
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
const status = classifyTerminalStatus({ aborted: ac.signal.aborted, error: err });
log.error({ taskId, agent, err: errMsg }, 'dispatcher: warm ACP error');
await sql`
UPDATE tasks
SET state = 'failed', ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId}
SET state = ${status}, ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId} AND state NOT IN ('cancelled', 'completed')
`.catch(() => {});
// F1 (OCE-001): finalize the streaming message (was left 'streaming').
await finalizeMessage(sessionId, chatId, assistantId, status, task.model);
// #10: turn crashed.
emitAgentStatus(sessionId, chatId, agent, 'error', 'crashed');
emitAgentStatus(sessionId, chatId, agent, status === 'cancelled' ? 'idle' : 'error', status === 'cancelled' ? 'cancelled' : 'crashed');
clearTaskCommands(taskId);
// No worktree cleanup (persistent); backend stays warm for the next turn.
}
@@ -1245,7 +1342,10 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
return;
}
const ac = new AbortController();
// F1: register the per-task abort controller so a Stop reaches this run.
const ac = taskControllers.register(taskId);
// F1: hoisted so the catch / abort short-circuit can finalize the streaming row.
let assistantId = '';
try {
await sql`
@@ -1267,7 +1367,7 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
VALUES (${sessionId}, ${chatId}, 'assistant', '', 'streaming', ${task.model}, clock_timestamp())
RETURNING id
`;
const assistantId = assistantMsg!.id;
assistantId = assistantMsg!.id;
// write-edit-robustness #4: pre-turn checkpoint of the persistent session
// worktree (best-effort; never breaks dispatch).
@@ -1376,6 +1476,18 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
await persistExternalAgentTurn(sql, assistantId, [...toolSnaps.values()], reasoningText);
// F1: abort short-circuit BEFORE the unconditional 'complete' write — fixes
// the warm success-path recording 'complete' on a Stop'd turn. The abort fired
// the SDK interrupt on the same query generator only (never killed the warm
// process), so the persistent worktree is kept and the backend stays warm.
if (ac.signal.aborted || stopping) {
await finalizeMessage(sessionId, chatId, assistantId, 'cancelled', task.model, assistantContent);
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
emitAgentStatus(sessionId, chatId, agent, 'idle', stopping ? 'shutdown' : 'cancelled');
clearTaskCommands(taskId);
return; // worktree persists (no cleanup); backend stays warm
}
// ctx_used/ctx_max from the SDK result (1M-aware) → the assistant message, so
// the ContextBar renders a real context-window fill for claude.
await sql`
@@ -1391,11 +1503,6 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
model: task.model,
} as WsFrame);
if (stopping) {
await sql`UPDATE tasks SET state = 'cancelled', ended_at = clock_timestamp() WHERE id = ${taskId}`;
return; // worktree persists (no cleanup); backend stays warm
}
// Diff the persistent worktree against its captured baseline and SUPERSEDE
// the session's prior pending row (latest-wins) — identical to opencode/ACP.
const diff = await diffWorktree(worktreePath, projectPath, {
@@ -1442,14 +1549,17 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
clearTaskCommands(taskId);
} catch (err) {
const errMsg = err instanceof Error ? err.message : String(err);
const status = classifyTerminalStatus({ aborted: ac.signal.aborted, error: err });
log.error({ taskId, agent, err: errMsg }, 'dispatcher: claude SDK error');
await sql`
UPDATE tasks
SET state = 'failed', ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId}
SET state = ${status}, ended_at = clock_timestamp(), output_summary = ${errMsg.slice(0, 500)}
WHERE id = ${taskId} AND state NOT IN ('cancelled', 'completed')
`.catch(() => {});
// F1 (OCE-001): finalize the streaming message (was left 'streaming').
await finalizeMessage(sessionId, chatId, assistantId, status, task.model);
// #10: turn crashed.
emitAgentStatus(sessionId, chatId, agent, 'error', 'crashed');
emitAgentStatus(sessionId, chatId, agent, status === 'cancelled' ? 'idle' : 'error', status === 'cancelled' ? 'cancelled' : 'crashed');
clearTaskCommands(taskId);
// No worktree cleanup (persistent); backend stays warm for the next turn.
}
@@ -1476,6 +1586,7 @@ export function createDispatcher(deps: Deps): { start(): void; stop(): Promise<v
}
return {
cancelExternalTask,
start() {
log.info('dispatcher: starting poll loop + tasks_new listener');

View File

@@ -0,0 +1,76 @@
import type { Sql } from '../db.js';
import type { WsFrame } from '@boocode/contracts/ws-frames';
export type TerminalMessageStatus = 'cancelled' | 'failed';
/**
* F1 (D-7) — decide the terminal status a Stop'd / errored external turn lands in.
*
* A user Stop (the per-task AbortController fired) or a thrown `AbortError` is a
* deliberate, non-error outcome → `'cancelled'`. A genuine thrown error → `'failed'`.
* Keeping the two distinct keeps the human-inbox / failure surfaces honest.
*
* Pure (no DB / IO) so the mapping is unit-testable in isolation.
*/
export function classifyTerminalStatus(opts: { aborted: boolean; error?: unknown }): TerminalMessageStatus {
if (opts.aborted) return 'cancelled';
if (opts.error instanceof Error && opts.error.name === 'AbortError') return 'cancelled';
return 'failed';
}
/**
* F1 (OCE-001 / OCE-002) — finalize a streaming assistant message into a terminal
* state and publish the matching `message_complete` frame.
*
* Idempotent via `WHERE status = 'streaming'`: a second call (a double-Stop, or an
* abort short-circuit followed by the catch block) updates zero rows and does NOT
* re-publish, so the frontend reducer settles the message exactly once. It also
* never clobbers a row that already finished cleanly (`complete`) — the abort that
* raced a clean finish is a no-op.
*
* Returns `true` iff this call performed the finalization (the row was still
* streaming); `false` if it was already terminal or the id is absent (the throw
* preceded the row's creation).
*/
export async function finalizeStreamingMessage(
sql: Sql,
publishFrame: (sessionId: string, frame: WsFrame) => void,
opts: {
sessionId: string;
chatId: string;
assistantId: string;
status: TerminalMessageStatus;
model: string | null;
/** Partial accumulated text to persist; omit to leave the row's content untouched. */
content?: string;
},
): Promise<boolean> {
const { sessionId, chatId, assistantId, status, model, content } = opts;
if (!assistantId) return false;
const rows =
content !== undefined
? await sql<{ id: string }[]>`
UPDATE messages
SET content = ${content}, status = ${status}, finished_at = clock_timestamp()
WHERE id = ${assistantId} AND status = 'streaming'
RETURNING id
`
: await sql<{ id: string }[]>`
UPDATE messages
SET status = ${status}, finished_at = clock_timestamp()
WHERE id = ${assistantId} AND status = 'streaming'
RETURNING id
`;
if (rows.length === 0) return false;
publishFrame(sessionId, {
type: 'message_complete',
message_id: assistantId,
chat_id: chatId,
model,
status,
} as WsFrame);
return true;
}

View File

@@ -29,6 +29,17 @@ interface ProjectPathRow {
path: string;
}
interface MessageRow {
id: string;
session_id: string;
chat_id: string | null;
role: string;
content: string;
status: string;
model: string | null;
created_at: Date;
}
function textResult(data: unknown) {
return { content: [{ type: 'text' as const, text: JSON.stringify(data, null, 2) }] };
}
@@ -189,6 +200,56 @@ export async function startMcpServer(sql: Sql): Promise<void> {
},
);
// 6. boocoder.view_session_history
server.tool(
'boocoder.view_session_history',
'Retrieve the most-recent N messages of a session chat transcript (role != system) from messages_with_parts, returned in chronological (oldest→newest) order',
{
session_id: z.string().describe('Session UUID'),
chat_id: z.string().optional().describe('Optional chat UUID — narrows to one chat tab'),
limit: z
.number()
.int()
.min(1)
.max(200)
.optional()
.describe('Max messages to return (default 50, max 200)'),
},
async (args) => {
const effectiveLimit = Math.min(args.limit ?? 50, 200);
let rows: MessageRow[];
if (args.chat_id) {
rows = await sql<MessageRow[]>`
SELECT id, session_id, chat_id, role, content, status, model, created_at
FROM (
SELECT id, session_id, chat_id, role, content, status, model, created_at
FROM messages_with_parts
WHERE session_id = ${args.session_id}
AND chat_id = ${args.chat_id}
AND role != 'system'
ORDER BY created_at DESC
LIMIT ${effectiveLimit}
) sub
ORDER BY created_at ASC
`;
} else {
rows = await sql<MessageRow[]>`
SELECT id, session_id, chat_id, role, content, status, model, created_at
FROM (
SELECT id, session_id, chat_id, role, content, status, model, created_at
FROM messages_with_parts
WHERE session_id = ${args.session_id}
AND role != 'system'
ORDER BY created_at DESC
LIMIT ${effectiveLimit}
) sub
ORDER BY created_at ASC
`;
}
return textResult({ session_id: args.session_id, count: rows.length, messages: rows });
},
);
// Connect via stdio
const transport = new StdioServerTransport();
await server.connect(transport);