// apps/vis/server/src/lib/task-store.ts // // Read-only reader for background tasks, persisted by the engine under each // spawning agent's homedir at `/tasks/.json` // (+ `tasks//output.log`). Main-agent reads may also receive the // legacy session root as a fallback. // // The visualizer never writes these files; it mirrors the engine's on-disk // layout (`packages/agent-core-v2/src/agent/task/persist.ts`) for reading only: // - the same `VALID_TASK_ID` guard, so a corrupt / hand-edited filename // cannot turn a log path into a traversal primitive; // - the same legacy snake_case → current camelCase normalization, so old // sessions list identically to how the CLI would list them. import { open, readdir, readFile } from 'node:fs/promises'; import { join } from 'node:path'; import type { BackgroundTaskInfo, BackgroundTaskStatus, } from './agent-record-types'; /** Task id format: `{prefix}-{8 chars of [0-9a-z]}`. Mirror of the engine's * `VALID_TASK_ID` (`agent/task/persist.ts`). Enforced before deriving any * output path so neither `../` nor a legacy `bg_` id can escape. */ const VALID_TASK_ID = /^[a-z0-9]+(?:-[a-z0-9]+)*-[0-9a-z]{8}$/; export function isSafeTaskId(id: string): boolean { return VALID_TASK_ID.test(id); } function tasksDirOf(agentDir: string): string { return join(agentDir, 'tasks'); } function taskOutputFile(agentDir: string, taskId: string): string { if (!VALID_TASK_ID.test(taskId)) { throw new Error(`Invalid task id: "${taskId}"`); } return join(tasksDirOf(agentDir), taskId, 'output.log'); } /** * Enumerate all persisted background tasks for a session, normalized to the * current `BackgroundTaskInfo` shape and sorted newest-first by start time. * * Silently skips: filenames that don't match `VALID_TASK_ID`, files that fail * to read/parse, and records that are neither the current nor the legacy * task shape — matching the engine's tolerant `listTasks`. */ export async function listBackgroundTasks( agentDir: string, fallbackDir?: string, ): Promise { const primary = await listBackgroundTasksAt(agentDir); const out = [...primary.tasks]; if (fallbackDir !== undefined) { const fallback = await listBackgroundTasksAt(fallbackDir); for (const task of fallback.tasks) { if (!primary.reservedIds.has(task.keyId)) out.push(task); } } // Newest first; tasks with no start time sort last. out.sort((a, b) => (b.task.startedAt ?? 0) - (a.task.startedAt ?? 0)); return out.map((entry) => entry.task); } interface ListedTask { keyId: string; task: BackgroundTaskInfo; } async function listBackgroundTasksAt( agentDir: string, ): Promise<{ reservedIds: Set; tasks: ListedTask[] }> { const dir = tasksDirOf(agentDir); let entries: import('node:fs').Dirent[]; try { entries = await readdir(dir, { withFileTypes: true }); } catch { return { reservedIds: new Set(), tasks: [] }; } const reservedIds = new Set(); const tasks: ListedTask[] = []; for (const entry of entries) { if (!entry.name.endsWith('.json')) continue; const id = entry.name.slice(0, -'.json'.length); if (!VALID_TASK_ID.test(id)) continue; reservedIds.add(id); if (!entry.isFile()) continue; let parsed: unknown; try { parsed = JSON.parse(await readFile(join(dir, entry.name), 'utf8')); } catch { continue; } if (!isReadablePersistedTask(parsed)) continue; const task = normalizePersistedTask(parsed); if (task === undefined || task.taskId !== id) continue; tasks.push({ keyId: id, task }); } return { reservedIds, tasks }; } export interface TaskOutputMetadata { exists: boolean; size: number; } /** Presence and byte size of a task's `output.log`. */ export async function taskOutputMetadata( agentDir: string, taskId: string, fallbackDir?: string, ): Promise { const handle = await openTaskOutput(agentDir, taskId, fallbackDir); if (handle === undefined) return { exists: false, size: 0 }; try { return { exists: true, size: (await handle.stat()).size }; } catch { return { exists: false, size: 0 }; } finally { await handle.close(); } } /** Byte size of a task's `output.log` (0 when absent, empty, or unreadable). */ export async function taskOutputSizeBytes( agentDir: string, taskId: string, fallbackDir?: string, ): Promise { return (await taskOutputMetadata(agentDir, taskId, fallbackDir)).size; } export interface TaskOutputWindow { /** Byte offset this window starts at (clamped to >= 0). */ offset: number; /** Byte offset immediately after this window — pass it as the next * `offset` to page forward. Exact (server-computed bytesRead), so paging * never drifts even if a window boundary splits a multi-byte char. */ nextOffset: number; /** Total byte size of the log on disk. */ size: number; /** UTF-8 decoded window content. */ content: string; /** True when this window reaches EOF. */ eof: boolean; } /** * Read a byte window of a task's `output.log`. * * Reads at most `maxBytes` bytes starting at byte `offset`. A window past EOF * is clamped to whatever remains; an offset at/after EOF yields empty content. * Mirrors the engine's `readTaskOutputBytes` so large logs page identically. */ export async function readTaskOutput( agentDir: string, taskId: string, offset: number, maxBytes: number, fallbackDir?: string, ): Promise { const start = Math.max(0, Math.trunc(offset)); const limit = Math.max(0, Math.trunc(maxBytes)); const handle = await openTaskOutput(agentDir, taskId, fallbackDir); if (handle === undefined) { return { offset: start, nextOffset: start, size: 0, content: '', eof: true }; } try { const size = (await handle.stat()).size; if (limit === 0 || start >= size) { return { offset: start, nextOffset: start, size, content: '', eof: start >= size }; } const length = Math.min(limit, size - start); const buffer = Buffer.allocUnsafe(length); const { bytesRead } = await handle.read(buffer, 0, length, start); const content = buffer.toString('utf-8', 0, bytesRead); const nextOffset = start + bytesRead; return { offset: start, nextOffset, size, content, eof: nextOffset >= size }; } catch { return { offset: start, nextOffset: start, size: 0, content: '', eof: true }; } finally { await handle.close(); } } async function openTaskOutput( agentDir: string, taskId: string, fallbackDir?: string, ): Promise> | undefined> { try { return await open(taskOutputFile(agentDir, taskId), 'r'); } catch (error) { if (!isMissingPath(error) || fallbackDir === undefined) return undefined; } try { return await open(taskOutputFile(fallbackDir, taskId), 'r'); } catch { return undefined; } } function isMissingPath(error: unknown): boolean { return ( typeof error === 'object' && error !== null && 'code' in error && (error as NodeJS.ErrnoException).code === 'ENOENT' ); } // ── normalization (ported from agent-core-v2/agent/task/persist.ts) ──────── type ReadablePersistedTask = Record; interface CurrentTaskBase { readonly taskId: string; readonly description: string; readonly status: BackgroundTaskStatus; readonly detached: boolean; readonly startedAt: number; readonly endedAt: number | null; readonly stopReason?: string; readonly terminalNotificationSuppressed?: boolean; readonly resumeReminded?: boolean; readonly timeoutMs?: number; } const CURRENT_TASK_STATUSES: ReadonlySet = new Set([ 'running', 'completed', 'failed', 'timed_out', 'killed', 'lost', ]); function normalizePersistedTask(task: ReadablePersistedTask): BackgroundTaskInfo | undefined { const current = isLegacyPersistedTask(task) ? legacyPersistedTaskToCurrent(task) : task; return decodeCurrentPersistedTask(current); } function decodeCurrentPersistedTask(task: ReadablePersistedTask): BackgroundTaskInfo | undefined { const base = decodeCurrentTaskBase(task); if (base === undefined) return undefined; switch (task['kind']) { case 'process': if ( typeof task['command'] !== 'string' || !isFiniteNumber(task['pid']) || !isNullableFiniteNumber(task['exitCode']) ) { return undefined; } return { ...base, kind: 'process', command: task['command'], pid: task['pid'], exitCode: task['exitCode'], parentToolCallId: optionalString(task['parentToolCallId']), }; case 'agent': return { ...base, kind: 'agent', agentId: optionalString(task['agentId']), subagentType: optionalString(task['subagentType']), parentToolCallId: optionalString(task['parentToolCallId']), model: optionalString(task['model']), thinkingEffort: optionalString(task['thinkingEffort']), stopCode: optionalString(task['stopCode']), }; case 'question': if (!isFiniteNumber(task['questionCount'])) return undefined; return { ...base, kind: 'question', questionCount: task['questionCount'], toolCallId: optionalString(task['toolCallId']), }; default: return undefined; } } function decodeCurrentTaskBase(task: ReadablePersistedTask): CurrentTaskBase | undefined { if ( typeof task['taskId'] !== 'string' || !VALID_TASK_ID.test(task['taskId']) || typeof task['description'] !== 'string' || !isCurrentTaskStatus(task['status']) || !isFiniteNumber(task['startedAt']) || !isNullableFiniteNumber(task['endedAt']) ) { return undefined; } return { taskId: task['taskId'], description: task['description'], status: task['status'], detached: optionalBoolean(task['detached']) ?? true, startedAt: task['startedAt'], endedAt: task['endedAt'], stopReason: optionalString(task['stopReason']), terminalNotificationSuppressed: optionalBoolean(task['terminalNotificationSuppressed']), resumeReminded: optionalBoolean(task['resumeReminded']), timeoutMs: optionalNumber(task['timeoutMs']), }; } function legacyPersistedTaskToCurrent( task: ReadablePersistedTask & { readonly task_id: string }, ): ReadablePersistedTask { const base: ReadablePersistedTask = { taskId: task.task_id, description: task['description'], status: legacyStatusToCurrent(task), detached: true, startedAt: task['started_at'], endedAt: task['ended_at'], stopReason: optionalNonEmptyString(task['stop_reason']), timeoutMs: optionalNumber(task['timeout_ms']), }; if (task.task_id.startsWith('agent-')) { return { ...base, kind: 'agent', agentId: optionalNonEmptyString(task['agent_id']), subagentType: optionalNonEmptyString(task['subagent_type']), }; } return { ...base, kind: 'process', command: task['command'], pid: task['pid'], exitCode: task['exit_code'], }; } function legacyStatusToCurrent(task: ReadablePersistedTask): unknown { if (task['status'] === 'awaiting_approval') return 'running'; if (task['status'] === 'failed' && task['timed_out'] === true) return 'timed_out'; return task['status']; } function isReadablePersistedTask(obj: unknown): obj is ReadablePersistedTask { return ( isRecord(obj) && (typeof obj['taskId'] === 'string' || typeof obj['task_id'] === 'string') ); } function isLegacyPersistedTask( task: ReadablePersistedTask, ): task is ReadablePersistedTask & { readonly task_id: string } { return typeof task['task_id'] === 'string'; } function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null; } function optionalNonEmptyString(value: unknown): string | undefined { if (typeof value !== 'string') return undefined; const trimmed = value.trim(); return trimmed.length > 0 ? trimmed : undefined; } function optionalString(value: unknown): string | undefined { return typeof value === 'string' ? value : undefined; } function optionalBoolean(value: unknown): boolean | undefined { return typeof value === 'boolean' ? value : undefined; } function optionalNumber(value: unknown): number | undefined { return isFiniteNumber(value) ? value : undefined; } function isFiniteNumber(value: unknown): value is number { return typeof value === 'number' && Number.isFinite(value); } function isNullableFiniteNumber(value: unknown): value is number | null { return value === null || isFiniteNumber(value); } function isCurrentTaskStatus(value: unknown): value is BackgroundTaskStatus { return ( typeof value === 'string' && CURRENT_TASK_STATUSES.has(value as BackgroundTaskStatus) ); }