/** * Background-job routes of the /sidebar JSON API ('jobs.output' / * 'jobs.kill'). The job LIST needs no route: it arrives through the * harness's `session/jobs` push mirror (`jobsBySession` in the sessions * list feed). The routes: * * - 'jobs.output' — REPLAYS the output the MODEL has read so far for one * job. The source is the owner session's own event log: `tool/call` rows * of `job_output` name the job via `arguments.job_id`, and the paired * `tool/result` rows carry the finalized content the model received. * Because the session store's in-memory log can lag the live append feed * after a host restart (the store session stays frozen at its * rehydration boundary), the plugin ALSO mirrors job_output events from * the live `session/event` feed and merges both sources (deduped by seq). * This touches NO DSH source: the model's `job_output` cursor is never * consumed, and the pane stays empty until the agent reads the job. * - 'jobs.kill' — the registry's stock `kill` (a pristine DSH API), * fenced by the owning session via the live agent caller. Absent registry * → 503, mirroring the settings routes' optional-service downgrade. */ import type { Context, SidebarSessionEvent } from './context-types.ts' import { requireString, SidebarError } from './wire.ts' /** The two background-job routes of the sidebar API. */ export interface SidebarJobsRoutes { /** The output the model has read so far for one job (event replay, capped). */ output(payload: unknown): { text: string; truncated: boolean; read: boolean } /** Request cancellation of one job (live jobs flip to stopping). */ kill(payload: unknown): { ok: true; outcome: 'requested' | 'already-finished' } } /** The 'tool/result' message envelope inside a session event's data. */ interface ToolResultMessageLike { source?: { kind?: unknown; callId?: unknown } content?: unknown } /** One 'tool-result' content block (the inner blocks carry the text). */ interface ToolResultBlockLike { type?: unknown content?: unknown isError?: unknown } /** * Extract the plain text of a finalized tool result: the text blocks inside * the 'tool-result' block, joined with newlines. Error results and * non-text blocks contribute nothing. */ function resultText(message: ToolResultMessageLike): string | undefined { if (!Array.isArray(message.content)) return undefined const parts: string[] = [] for (const block of message.content) { if (block === null || typeof block !== 'object') continue const candidate = block as ToolResultBlockLike if (candidate.type !== 'tool-result') continue const inner = candidate.content if (!Array.isArray(inner)) continue for (const item of inner) { if (item === null || typeof item !== 'object') continue const textItem = item as { type?: unknown; text?: unknown } if (textItem.type === 'text' && typeof textItem.text === 'string') { parts.push(textItem.text) } } } return parts.length > 0 ? parts.join('\n') : undefined } /** Whether a tool/result is an error result (the inner block's isError flag). */ function resultIsError(message: ToolResultMessageLike): boolean { if (!Array.isArray(message.content)) return false return message.content.some((block) => { if (block === null || typeof block !== 'object') return false return (block as ToolResultBlockLike).type === 'tool-result' && (block as ToolResultBlockLike).isError === true }) } /** Whether a job_output result carries no new output — the controller's * model-facing "(no new output)" body, noise for the human pane. */ function isNoNewOutput(text: string): boolean { return text.startsWith('(no new output)') } /** One compact job_output trace (a tool/call or its paired tool/result). */ interface JobOutputTrace { seq: number kind: 'call' | 'result' /** The tool call identity pairing the two rows. */ callId: string /** tool/call: the job id parsed from the model arguments. */ jobId?: string /** tool/result: the finalized text the model received. */ text?: string /** tool/result: whether the result was an error (read counts, text skipped). */ isError?: boolean } /** Extract the job_output trace of one raw session event (undefined = unrelated). */ function traceOf(event: SidebarSessionEvent): JobOutputTrace | undefined { if (event.type === 'tool/call') { const data = event.data as { name?: unknown; callId?: unknown; arguments?: unknown } if (data.name !== 'job_output' || typeof data.callId !== 'string') return undefined let jobId: string | undefined try { const args = JSON.parse(typeof data.arguments === 'string' ? data.arguments : '') as { job_id?: unknown } if (typeof args.job_id === 'string') jobId = args.job_id } catch { // Malformed model arguments: not a job_output pair. } if (jobId === undefined) return undefined return { seq: event.seq, kind: 'call', callId: data.callId, jobId } } if (event.type === 'tool/result') { const message = (event.data as { message?: unknown }).message as ToolResultMessageLike | undefined if (message === undefined) return undefined const callId = message.source?.callId if (typeof callId !== 'string') return undefined return { seq: event.seq, kind: 'result', callId, text: resultText(message), isError: resultIsError(message), } } return undefined } /** Per-session cap of mirrored live traces (a bounded, lossy ring). */ const MIRROR_MAX_ENTRIES = 200 /** * The live job_output mirror: subscribes to the session append feed and * caches the job_output traces the session store's own log can lag behind * (after a host restart the store session stays frozen at its rehydration * boundary, so `session.events` misses everything appended since — the very * reads the pane exists to show). Zero DSH writes: the api-proxy pushes the * same feed to browsers. */ function createJobOutputMirror(ctx: Context): { entries(sessionId: string): readonly JobOutputTrace[] } { const perSession = new Map() // tool/call identities per session, so unrelated tool/result rows are // never cached (only job_output results pair with a cached call). const callIds = new Map>() if (typeof ctx.on !== 'function') { // Test doubles without the event API degrade to seed-only replay. return { entries: () => [] } } const dispose = ctx.on('session/event', (session, event) => { const sessionId = (session as { id?: unknown } | null)?.id if (typeof sessionId !== 'string') return if (event.type === 'tool/call') { const trace = traceOf(event) if (trace?.kind !== 'call') return let ids = callIds.get(sessionId) if (ids === undefined) callIds.set(sessionId, ids = new Set()) ids.add(trace.callId) push(sessionId, trace) } else if (event.type === 'tool/result') { const trace = traceOf(event) if (trace?.kind !== 'result') return if (!callIds.get(sessionId)?.has(trace.callId)) return push(sessionId, trace) } }) ctx.effect(() => dispose, 'dsh-better-sidebar: job-output event mirror') const push = (sessionId: string, trace: JobOutputTrace): void => { let list = perSession.get(sessionId) if (list === undefined) perSession.set(sessionId, list = []) list.push(trace) if (list.length > MIRROR_MAX_ENTRIES) { const removed = list.splice(0, list.length - MIRROR_MAX_ENTRIES) const ids = callIds.get(sessionId) if (ids !== undefined) { for (const entry of removed) { if (entry.kind === 'call') ids.delete(entry.callId) } if (ids.size === 0) callIds.delete(sessionId) } } } return { entries: (sessionId) => perSession.get(sessionId) ?? [] } } /** * Build the jobs routes bound to the plugin context. `output` merges the * owner session's own event log with the live job_output mirror; `kill` * reads the jobs/agents services lazily and degrades to a 503 when the * deployment lacks the registry. * @param ctx - host plugin context. * @param outputLimit - response cap for one output replay in bytes; longer * texts are sliced and flagged `truncated` (mirrors the fs.read cap). */ export function buildJobsApi(ctx: Context, outputLimit: number): SidebarJobsRoutes { const jobs = ctx.get('jobs') const agents = ctx.get('agents') const mirror = createJobOutputMirror(ctx) /** The live caller whose session id the registry fence compares against. */ const callerOf = (sessionId: string) => agents?.get(sessionId) /** Registry refusals become a 404 job-error; unknown and foreign ids are indistinguishable. */ const registryError = (error: unknown): SidebarError => new SidebarError('job-error', error instanceof Error ? error.message : String(error), 404) return { output(payload) { const sessionId = requireString(payload, 'sessionId') const id = requireString(payload, 'id') // Merge the store's event log (durable seed + whatever it received) // with the live mirror, deduped by seq — a trace never double-counts. const bySeq = new Map() for (const event of ctx.sessions.get(sessionId)?.snapshotEvents() ?? []) { const trace = traceOf(event) if (trace !== undefined) bySeq.set(trace.seq, trace) } for (const trace of mirror.entries(sessionId)) bySeq.set(trace.seq, trace) // Pair calls with results in seq order: the model's reads, oldest first. const jobOf = new Map() const parts: string[] = [] let read = false for (const trace of [...bySeq.values()].sort((left, right) => left.seq - right.seq)) { if (trace.kind === 'call') { if (trace.jobId !== undefined) jobOf.set(trace.callId, trace.jobId) } else if (jobOf.get(trace.callId) === id) { read = true if (trace.isError !== true && trace.text !== undefined && !isNoNewOutput(trace.text)) { parts.push(trace.text) } } } const text = parts.join('\n') return { text: text.length > outputLimit ? text.slice(0, outputLimit) : text, truncated: text.length > outputLimit, read, } }, kill(payload) { if (jobs === undefined) { throw new SidebarError('job-error', 'the background-job registry is not mounted in this deployment', 503) } const sessionId = requireString(payload, 'sessionId') const id = requireString(payload, 'id') const record = payload as { reason?: unknown } | null const reason = typeof record?.reason === 'string' && record.reason !== '' ? record.reason : 'user requested via sidebar' try { return { ok: true, outcome: jobs.kill(id, callerOf(sessionId), reason) } } catch (error) { throw registryError(error) } }, } }