diff --git a/src/commands/connect.ts b/src/commands/connect.ts index a6bccba..4532307 100644 --- a/src/commands/connect.ts +++ b/src/commands/connect.ts @@ -1,17 +1,14 @@ import type { Command } from 'commander' -import { createInterface } from 'node:readline' -import { ApiClient, ApiError } from '../lib/api' +import React from 'react' +import { render } from 'ink' +import { ApiClient } from '../lib/api' import { requireToken, resolveApiBase, resolveAppBase } from '../lib/config' import { runAction } from '../lib/output' -import { formatStepLine } from '../lib/steps' +import { eventToItems, type CCEvent, type TranscriptItem } from '../lib/events' import { sessionUrl } from '../lib/urls' -import { - resolveWsBase, - streamSession, - StreamUnavailableError, - type StreamFrame, -} from '../lib/ws' -import type { AgentSession } from '../lib/types' +import { resolveWsBase } from '../lib/ws' +import { ConnectApp } from '../ui/ConnectApp' +import type { AgentSession, AgentStep } from '../lib/types' // `agent session connect [sessionId]` — the terminal window into a cloud // session (documents/eng/SESSION_IDE.md §2.6, in the ellipsis monorepo). @@ -106,160 +103,52 @@ export async function runConnect( const reason = readOnly && c.canSend ? 'read-only (--no-input) — following without the composer' : c.reason - console.log(`session: ${session.id} (${session.status})`) - if (session.agent_config_id) console.log(`config: ${session.agent_config_id}`) - console.log(`url: ${sessionUrl(resolveAppBase(), me.customer_login, sessionId)}`) + // A compact header printed once above the live transcript. + console.log(`${session.id} · ${session.status}`) + console.log(sessionUrl(resolveAppBase(), me.customer_login, sessionId)) + if (session.agent_config_id) console.log(`config ${session.agent_config_id}`) if (reason) console.log(reason) - - if (showSteps) { - const steps = await api.getAgentSessionSteps(sessionId) - if (steps.length > 0) { - const ordered = [...steps].sort( - (a, b) => a.created_at.localeCompare(b.created_at) || a.step_index - b.step_index, - ) - console.log('') - for (const step of ordered) console.log(formatStepLine(step)) - } - } - - if (!canSend) { - // Watch-only: follow the live stream to its terminal frame and exit. - console.log('') - await followOnce(token, sessionId, wsBase, 0, (line) => console.log(line)) - return + if (canSend) { + console.log('type to send · /stop ends the turn · /exit or Ctrl+C detaches') } - console.log('') - console.log( - '── connected: type to send (delivered at the next turn boundary), ' + - '/stop ends the current turn, Ctrl+C detaches ──', + // Fetch the stored steps to seed the transcript (unless --no-steps) and to + // set the live-refresh cursor, so the live updates only append what's new. + // The step `data` is the same Claude Code event shape the UI renders live. + const steps = await api.getAgentSessionSteps(sessionId) + const initialMaxStepIndex = steps.reduce((m, s) => Math.max(m, s.step_index), -1) + const initialItems = showSteps ? stepsToItems(steps) : [] + + const app = render( + React.createElement(ConnectApp, { + api, + token, + sessionId, + wsBase, + canSend, + initialItems, + // Always advance the cursor past existing steps: --no-steps skips + // *rendering* history, not re-streaming it live. + initialMaxStepIndex, + initialStatus: session.status, + }), ) + await app.waitUntilExit() - const rl = createInterface({ input: process.stdin, output: process.stdout, prompt: '> ' }) - const abort = new AbortController() - // The live-output cursor, carried across stream re-attaches so an idle→wake - // transition (the previous execution's stream ends with `done`) resumes - // without dropping or repeating frames. - let afterSeq = 0 - let streaming = false - - // Print a line "above" the composer: clear the prompt line, write, redraw - // the prompt with whatever the user had typed. - const printAbove = (line: string): void => { - process.stdout.write('\r\x1b[K') - console.log(line) - rl.prompt(true) - } - - // Follow the session's live output until the current execution finishes. - // A keyed session going terminal is NOT the end of the conversation — it - // idles out between turns — so we report it quietly and re-attach on the - // next send instead of exiting. - const pump = (): void => { - if (streaming) return - streaming = true - void followOnce(token, sessionId, wsBase, afterSeq, printAbove, abort.signal, (seq) => { - afterSeq = Math.max(afterSeq, seq) - }) - .then((ended) => { - if (ended && !abort.signal.aborted) { - printAbove('· agent idle — a message wakes it') - } - }) - .catch((err: unknown) => { - if (err instanceof StreamUnavailableError) { - printAbove(`· live stream unavailable (${err.message}) — messages still send`) - } else { - printAbove(`✗ stream error: ${(err as Error).message}`) - } - }) - .finally(() => { - streaming = false - }) - } - pump() - - const detach = (): void => { - abort.abort() - rl.close() + if (canSend) { console.log('\ndetached — the session keeps running (reconnect with the same command)') } - - rl.on('SIGINT', detach) - rl.on('line', (raw) => { - const text = raw.trim() - if (!text) { - rl.prompt() - return - } - if (text === '/exit' || text === '/quit') { - detach() - return - } - void (async () => { - try { - if (text === '/stop') { - const s = await api.stopAgentSession(sessionId) - printAbove(`✓ stop requested (${s.status}) — the conversation survives`) - return - } - await api.sendSessionMessage(sessionId, text) - printAbove('· sent — delivered at the next turn boundary') - pump() // re-attach if the previous execution had ended - } catch (err) { - // 409s carry human-readable reasons (closed, budget floor); show as-is. - printAbove(`✗ ${err instanceof ApiError ? err.detail : (err as Error).message}`) - } - })() - }) - rl.prompt() - - // Keep the process alive until the user detaches. - await new Promise((resolve) => rl.on('close', resolve)) } -// One streaming attach: renders frames as display lines until the current -// execution's terminal frame. Resolves true when the stream ended normally -// (terminal `done`), false on abort. Collapses repeated status frames (the -// server's keepalive) exactly like `session get --watch`. -async function followOnce( - token: string, - sessionId: string, - wsBase: string, - afterSeq: number, - print: (line: string) => void, - signal?: AbortSignal, - onSeq?: (seq: number) => void, -): Promise { - let lastStatus: string | undefined - const onFrame = (frame: StreamFrame): void => { - if (typeof frame.seq === 'number') onSeq?.(frame.seq) - switch (frame.type) { - case 'status': - if (frame.status !== lastStatus) { - lastStatus = frame.status - print(`· ${frame.status}`) - } - break - case 'stdout': - case 'stderr': - if (frame.data) { - for (const line of frame.data.split('\n')) { - if (line.trim()) print(line) - } - } - break - case 'error': - print(`✗ ${frame.message ?? frame.data ?? 'stream error'}`) - break - case 'done': - break // reported by the caller - } - } - const outcome = await streamSession({ token, sessionId, wsBase, afterSeq, onFrame, signal }) - if (outcome.type === 'error') { - print(`✗ ${outcome.message}`) - return true +// Flatten stored steps into transcript items, in chronological order — the +// stored `data` is the same Claude Code event shape as the live stream. +function stepsToItems(steps: AgentStep[]): TranscriptItem[] { + const ordered = [...steps].sort( + (a, b) => a.created_at.localeCompare(b.created_at) || a.step_index - b.step_index, + ) + const items: TranscriptItem[] = [] + for (const step of ordered) { + items.push(...eventToItems(step.data as CCEvent, `s${step.step_index}`)) } - return outcome.type === 'done' + return items } diff --git a/src/commands/session.tsx b/src/commands/session.tsx index fac73fe..9a035d7 100644 --- a/src/commands/session.tsx +++ b/src/commands/session.tsx @@ -219,9 +219,7 @@ export function registerSession(program: Command): void { const session = await api.startAgentSession(req) if (opts.connect) { - console.log(`✓ started session ${session.id} (${session.status})`) - await printSessionUrl(api, session.id) - await startConnect(api, session.id) + await startConnect(api, session) return } @@ -754,33 +752,62 @@ export function registerSession(program: Command): void { // so there is nothing to connect to. const CONNECT_READY_TIMEOUT_SECONDS = 120 -export async function startConnect(api: ApiClient, sessionId: string): Promise { - process.stdout.write('waiting for the sandbox') - const deadline = Date.now() + CONNECT_READY_TIMEOUT_SECONDS * 1000 - for (;;) { - const s = await api.getAgentSession(sessionId) - if (s.status === 'running') break - if (TERMINAL_STATUSES.has(s.status)) { - const reason = s.status_reason ? ` — ${s.status_reason}` : '' - console.log(`\nsession ${s.status} before it became connectable${reason}`) - process.exitCode = 1 - return - } - if (Date.now() > deadline) { - console.log( - `\ntimed out waiting for the sandbox; once it is running, connect with:\n` + - ` agent session connect ${sessionId}`, - ) - process.exitCode = 1 - return +export async function startConnect(api: ApiClient, session: AgentSession): Promise { + const sessionId = session.id + // The sandbox may already be live for a warm session; otherwise show an + // animated wait (the transient line clears before the connect UI takes over, + // so the session id/url aren't printed twice). A terminal status before + // RUNNING means the session never got a sandbox (preflight/budget gate). + if (session.status !== 'running') { + const stop = startWaitSpinner(`starting ${sessionId} — waiting for the sandbox`) + const deadline = Date.now() + CONNECT_READY_TIMEOUT_SECONDS * 1000 + for (;;) { + const s = await api.getAgentSession(sessionId) + if (s.status === 'running') break + if (TERMINAL_STATUSES.has(s.status)) { + const reason = s.status_reason ? ` — ${s.status_reason}` : '' + stop(`session ${s.status} before it became connectable${reason}`) + process.exitCode = 1 + return + } + if (Date.now() > deadline) { + stop( + `timed out waiting for the sandbox; once it is running, connect with:\n` + + ` agent session connect ${sessionId}`, + ) + process.exitCode = 1 + return + } + await sleep(1000) } - process.stdout.write('.') - await sleep(1000) + stop() } - process.stdout.write(' ready\n') await runConnect(sessionId, true) } +// A single-line braille spinner for a pre-connect wait. `stop(final?)` clears +// the animated line and optionally prints a final message in its place. On a +// non-TTY (piped/CI) it prints one static line instead of animating. +function startWaitSpinner(label: string): (final?: string) => void { + if (!process.stdout.isTTY) { + console.log(`${label}…`) + return (final) => { + if (final) console.log(final) + } + } + const frames = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'] + let i = 0 + process.stdout.write('\x1b[?25l') // hide cursor + const draw = () => process.stdout.write(`\r\x1b[K${frames[i++ % frames.length]} ${label}…`) + draw() + const timer = setInterval(draw, 80) + return (final) => { + clearInterval(timer) + process.stdout.write('\r\x1b[K\x1b[?25h') // clear the line, restore cursor + if (final) console.log(final) + } +} + // `--watch` entry point: stream the session's output live over WebSocket, and // fall back to REST status polling if streaming is unavailable (e.g. a // backend without the endpoint). Identical UX either way — the same flag diff --git a/src/lib/events.ts b/src/lib/events.ts new file mode 100644 index 0000000..688a2fe --- /dev/null +++ b/src/lib/events.ts @@ -0,0 +1,251 @@ +// Turn the semantic event stream `agent session connect` receives into +// structured, renderable transcript items — the parsing/shaping layer behind +// the Claude-Code-like connect UI (src/ui/ConnectApp.tsx). +// +// The session WebSocket relays the agent's Claude Code stream-json: one JSON +// object per line, carried in `stdout` frames (see docs/RUN_STREAMING_SPEC.md +// and src/lib/ws.ts). Each object is a `CCEvent` — a discriminated union of +// assistant turns (text / thinking / tool_use blocks), user turns (tool +// results, or a human message), a system init, and a final result. This module +// is pure (no ANSI, no Ink) so it can be unit-tested directly; colours and +// layout live in the UI component. + +import { oneLine } from './steps' + +// A loosely-typed content block of a Claude Code message. We only read the +// fields we display and never interpret the rest of the payload. +export interface CCContentBlock { + type?: string + text?: string + thinking?: string + id?: string + name?: string + input?: Record + content?: unknown + tool_use_id?: string + is_error?: boolean +} + +// One Claude Code stream-json event. `message.content` is a string or a list +// of blocks; `result` (with cost/duration) caps a turn. Everything the CLI +// doesn't render is passed through untyped. +export interface CCEvent { + type?: string + subtype?: string + message?: { role?: string; content?: unknown } + result?: string + duration_ms?: number + total_cost_usd?: number + num_turns?: number + is_error?: boolean + model?: string + cwd?: string + [key: string]: unknown +} + +// A single line of the rendered transcript. `kind` selects the colour/glyph in +// the UI; `gutter` is the leading marker (● tool, ⎿ result, › you, ✻ thinking). +// `spaceBefore` opens a blank line above to separate message-level blocks; +// grouped sub-lines (a tool's result under its call) set it false so they hug. +export type ItemKind = + | 'assistant' + | 'thinking' + | 'tool' + | 'tool_result' + | 'summary' + | 'system' + | 'user' + | 'notice' + | 'error' + +export interface TranscriptItem { + key: string + kind: ItemKind + text: string + // Secondary, dimmed text shown after `text` on the same logical block (a + // tool call's argument summary, a result's body). + detail?: string + gutter?: string + spaceBefore?: boolean + isError?: boolean +} + +// Reassembles the byte chunks of `stdout`/`stderr` frames into whole lines: a +// single JSON event can be split across frames, so we only emit a line once its +// terminating newline has arrived. `flush()` releases any trailing partial at +// stream end. +export class LineBuffer { + private buf = '' + + push(chunk: string): string[] { + this.buf += chunk + const parts = this.buf.split('\n') + this.buf = parts.pop() ?? '' + return parts + } + + flush(): string[] { + const rest = this.buf.trim() + this.buf = '' + return rest ? [rest] : [] + } +} + +// Parse one relayed line into a CCEvent, or null if it isn't a JSON object (a +// blank keepalive line, or plain non-event text the caller renders verbatim). +export function parseEventLine(line: string): CCEvent | null { + const trimmed = line.trim() + if (!trimmed) return null + try { + const value = JSON.parse(trimmed) as unknown + if (value && typeof value === 'object' && !Array.isArray(value)) { + return value as CCEvent + } + } catch { + // Not JSON — the caller shows it as raw text. + } + return null +} + +// Claude-Code-style one-line summary of a tool call's arguments: the salient +// field for the common tools (a path, a command, a pattern), else compact JSON. +export function summarizeToolInput(name: string, input: unknown): string { + const args = (input && typeof input === 'object' ? input : {}) as Record + const str = (v: unknown): string | undefined => (typeof v === 'string' ? v : undefined) + const tool = name.toLowerCase() + + const path = str(args.file_path) ?? str(args.path) ?? str(args.notebook_path) + if (['read', 'write', 'edit', 'multiedit', 'notebookedit'].includes(tool) && path) { + return oneLine(path, 100) + } + if (tool === 'bash' && str(args.command)) return oneLine(str(args.command)!, 100) + if ((tool === 'grep' || tool === 'glob') && str(args.pattern)) { + const where = str(args.path) ?? str(args.glob) + return oneLine(str(args.pattern)! + (where ? ` in ${where}` : ''), 100) + } + if ((tool === 'task' || tool === 'agent') && str(args.description)) { + return oneLine(str(args.description)!, 100) + } + if (tool === 'webfetch' && str(args.url)) return oneLine(str(args.url)!, 100) + if (tool === 'websearch' && str(args.query)) return oneLine(str(args.query)!, 100) + + const keys = Object.keys(args) + if (keys.length === 0) return '' + return oneLine(JSON.stringify(args), 100) +} + +// Flatten a message's `content` to a block list (a bare string becomes one +// text block), so assistant and user turns share one iteration path. +function blocksOf(content: unknown): CCContentBlock[] { + if (typeof content === 'string') return [{ type: 'text', text: content }] + if (Array.isArray(content)) return content as CCContentBlock[] + return [] +} + +// Best-effort display text for a tool_result block: its content is a string, a +// list of text blocks, or arbitrary JSON. Whitespace is preserved (results are +// shown as a small indented body), only trimmed at the ends. +function toolResultText(block: CCContentBlock): string { + const c = block.content + if (typeof c === 'string') return c.trim() + if (Array.isArray(c)) { + const parts: string[] = [] + for (const inner of c as CCContentBlock[]) { + if (typeof inner.text === 'string') parts.push(inner.text) + else parts.push(JSON.stringify(inner)) + } + return parts.join('\n').trim() + } + if (c === undefined || c === null) return '' + return JSON.stringify(c) +} + +// Render a whole event into transcript items (usually one, sometimes several +// for a multi-block assistant turn). `keyBase` makes React keys unique across +// the stream; blank when the event carries nothing worth showing. +export function eventToItems(event: CCEvent, keyBase: string): TranscriptItem[] { + const items: TranscriptItem[] = [] + const push = (item: Omit): void => { + items.push({ ...item, key: `${keyBase}:${items.length}` }) + } + const type = event.type + + // A final result event: a dim one-liner capping the turn (cost + duration). + if (type === 'result') { + const bits: string[] = [] + if (typeof event.duration_ms === 'number') bits.push(`${(event.duration_ms / 1000).toFixed(1)}s`) + if (typeof event.total_cost_usd === 'number') bits.push(`$${event.total_cost_usd.toFixed(2)}`) + const label = event.is_error ? 'turn ended with an error' : 'turn complete' + push({ + kind: 'summary', + text: bits.length ? `${label} · ${bits.join(' · ')}` : label, + spaceBefore: true, + isError: event.is_error, + }) + return items + } + + // System init: a compact, dim "session started" line (model / cwd). + if (type === 'system') { + const bits: string[] = [] + if (typeof event.model === 'string') bits.push(event.model) + if (typeof event.cwd === 'string') bits.push(event.cwd) + push({ + kind: 'system', + text: bits.length ? `session started · ${bits.join(' · ')}` : 'session started', + spaceBefore: true, + }) + return items + } + + const blocks = blocksOf(event.message?.content) + + // A user turn is either tool results (grouped under the calls above) or a + // human message injected into the conversation. + if (type === 'user') { + for (const block of blocks) { + if (block.type === 'tool_result') { + const body = toolResultText(block) + push({ + kind: 'tool_result', + gutter: '⎿', + text: body || '(no output)', + spaceBefore: false, + isError: block.is_error, + }) + } else if (typeof block.text === 'string' && block.text.trim()) { + push({ kind: 'user', gutter: '›', text: block.text.trim(), spaceBefore: true }) + } + } + return items + } + + // Assistant turn (and any other event carrying message content): text, + // thinking, and tool calls, in order. + for (const block of blocks) { + if (block.type === 'thinking' && typeof block.thinking === 'string' && block.thinking.trim()) { + push({ kind: 'thinking', gutter: '✻', text: block.thinking.trim(), spaceBefore: true }) + } else if (block.type === 'tool_use') { + const name = block.name ?? 'tool' + const summary = summarizeToolInput(name, block.input) + push({ + kind: 'tool', + gutter: '●', + text: name, + detail: summary ? `(${summary})` : undefined, + spaceBefore: true, + }) + } else if (typeof block.text === 'string' && block.text.trim()) { + push({ kind: 'assistant', text: block.text.trim(), spaceBefore: true }) + } + } + return items +} + +// Clamp a multi-line body to `maxLines`, appending a dim "+N lines" marker when +// truncated — used for tool-result bodies so a huge file read stays compact. +export function clampLines(text: string, maxLines: number): { body: string; more: number } { + const lines = text.split('\n') + if (lines.length <= maxLines) return { body: text, more: 0 } + return { body: lines.slice(0, maxLines).join('\n'), more: lines.length - maxLines } +} diff --git a/src/ui/ConnectApp.tsx b/src/ui/ConnectApp.tsx new file mode 100644 index 0000000..de5cd17 --- /dev/null +++ b/src/ui/ConnectApp.tsx @@ -0,0 +1,435 @@ +import React, { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { Box, Text, useApp, useInput, useStdin } from 'ink' +import Spinner from 'ink-spinner' +import { ApiClient, ApiError } from '../lib/api' +import { streamSession, StreamUnavailableError, type StreamFrame } from '../lib/ws' +import { clampLines, eventToItems, type CCEvent, type ItemKind, type TranscriptItem } from '../lib/events' + +// The interactive `agent session connect` UI, modelled on Claude Code: a +// committed transcript that groups tool calls with their results and spaces +// messages apart, above a live footer with an animated status spinner and a +// composer that echoes what you send. Rendering shape lives in lib/events.ts +// (pure); this component owns the data flow, the composer, and the colours. +// +// Data flow: the session WebSocket relays plain *text* (not structured events), +// so it can't be grouped into tool calls / results. Instead we render from the +// structured steps API (GET /v1/sessions/{id}/steps, whose `data` is the full +// Claude Code event), and use the socket only as a low-latency "something +// changed" wake plus status source. A slow poll backs it up when the socket is +// unavailable. This mirrors how the dashboard re-reads steps on a notify. + +export interface ConnectAppProps { + api: ApiClient + token: string + sessionId: string + wsBase: string + // Keyed, open sessions accept messages (show the composer); single-shot / + // closed / --no-input sessions follow read-only and exit when the stream ends. + canSend: boolean + initialItems: TranscriptItem[] + // The highest step_index already rendered into initialItems, so live refreshes + // only append steps newer than what's on screen. + initialMaxStepIndex: number + initialStatus: string +} + +// Statuses in which the agent is actively producing output — drives the spinner. +function isWorkingStatus(status: string): boolean { + return ['scheduled', 'creating_sandbox', 'running', 'retrying'].includes(status) +} + +export function ConnectApp(props: ConnectAppProps): React.ReactElement { + const { api, token, sessionId, wsBase, canSend } = props + const { exit } = useApp() + const { isRawModeSupported } = useStdin() + + const [items, setItems] = useState(props.initialItems) + const [status, setStatus] = useState(props.initialStatus) + const [working, setWorking] = useState(isWorkingStatus(props.initialStatus)) + const [elapsed, setElapsed] = useState(0) + const [notice, setNotice] = useState(null) + const [input, setInput] = useState('') + // ctrl+r toggles full vs. collapsed tool output across the whole transcript. + const [expanded, setExpanded] = useState(false) + // Messages you've sent that the agent hasn't picked up yet — shown as a + // queued region below the composer, exactly like Claude Code. Each moves into + // the transcript when the backend relays it as a user turn, or when the turn + // idles (whichever first), so a send is never lost. + const [queued, setQueued] = useState([]) + + // Live-flow state that must survive re-renders without triggering them. + const afterSeq = useRef(0) + const streaming = useRef(false) + const abort = useRef(new AbortController()) + const lastStatus = useRef(props.initialStatus) + const keyCounter = useRef(0) + // The highest step_index committed to the transcript, and the refresh + // in-flight / re-run guards so overlapping wakes don't double-append. + const maxStep = useRef(props.initialMaxStepIndex) + const refreshing = useRef(false) + const pendingRefresh = useRef(false) + const refreshTimer = useRef | null>(null) + // A mirror of `queued` for async callbacks, and the texts already committed + // locally so the backend's later echo of the same send is dropped. + const queuedRef = useRef([]) + const flushed = useRef([]) + useEffect(() => { + queuedRef.current = queued + }, [queued]) + + // Append items, promoting a queued send when its user turn arrives and + // dropping the backend's echo of a send we already committed locally. + const append = useCallback((incoming: TranscriptItem[]): void => { + const commit: TranscriptItem[] = [] + for (const it of incoming) { + if (it.kind === 'user') { + if (queuedRef.current.includes(it.text)) { + setQueued((prev) => { + const j = prev.indexOf(it.text) + return j < 0 ? prev : [...prev.slice(0, j), ...prev.slice(j + 1)] + }) + } else { + const fi = flushed.current.indexOf(it.text) + if (fi >= 0) { + flushed.current.splice(fi, 1) + continue + } + } + } + commit.push(it) + } + if (commit.length) setItems((prev) => [...prev, ...commit]) + }, []) + + // Pull the structured steps and append any newer than what's on screen. This + // is the sole source of transcript content — the socket only decides *when* + // to call it. Guarded so concurrent wakes coalesce into one trailing refresh. + const refreshSteps = useCallback(async (): Promise => { + if (refreshing.current) { + pendingRefresh.current = true + return + } + refreshing.current = true + try { + const steps = await api.getAgentSessionSteps(sessionId) + const ordered = [...steps].sort( + (a, b) => a.created_at.localeCompare(b.created_at) || a.step_index - b.step_index, + ) + const fresh = ordered.filter((st) => st.step_index > maxStep.current) + if (fresh.length) { + for (const st of fresh) maxStep.current = Math.max(maxStep.current, st.step_index) + append(fresh.flatMap((st) => eventToItems(st.data as CCEvent, `s${st.step_index}`))) + } + } catch { + // Transient fetch failure — the next wake/poll retries. + } finally { + refreshing.current = false + if (pendingRefresh.current) { + pendingRefresh.current = false + void refreshSteps() + } + } + }, [api, append, sessionId]) + + // Coalesce bursts of wakes into one refresh a beat later. + const scheduleRefresh = useCallback((): void => { + if (refreshTimer.current) return + refreshTimer.current = setTimeout(() => { + refreshTimer.current = null + void refreshSteps() + }, 250) + }, [refreshSteps]) + + // The socket carries status transitions (drive the spinner) and acts as a + // wake: any frame means new steps may exist, so schedule a refresh. + const handleFrame = useCallback( + (frame: StreamFrame): void => { + if (typeof frame.seq === 'number') afterSeq.current = Math.max(afterSeq.current, frame.seq) + if (frame.type === 'status' && frame.status && frame.status !== lastStatus.current) { + lastStatus.current = frame.status + setStatus(frame.status) + setWorking(isWorkingStatus(frame.status)) + } + if (frame.type === 'error') { + setNotice(frame.message ?? frame.data ?? 'stream error') + return + } + scheduleRefresh() + }, + [scheduleRefresh], + ) + + // Keep the socket attached across reconnects/resume. A keyed session going + // terminal is not the end of the conversation — it idles between turns — so + // we refresh once more, report idle, and re-attach on the next send. + // Watch-only sessions exit when the stream ends. + const pump = useCallback((): void => { + if (streaming.current || abort.current.signal.aborted) return + streaming.current = true + streamSession({ + token, + sessionId, + wsBase, + afterSeq: afterSeq.current, + onFrame: handleFrame, + signal: abort.current.signal, + }) + .then(async (outcome) => { + if (abort.current.signal.aborted) return + await refreshSteps() // pull the final steps of the turn before settling + setWorking(false) + if (outcome.type === 'error') setNotice(`stream error: ${outcome.message}`) + if (canSend) { + if (outcome.type === 'done') setNotice('agent idle — send a message to wake it') + } else { + exit() + } + }) + .catch((err: unknown) => { + if (abort.current.signal.aborted) return + const msg = + err instanceof StreamUnavailableError + ? `live stream unavailable (${err.message}) — polling for updates` + : `stream error: ${(err as Error).message}` + setNotice(msg) + // No socket: the poll below keeps the transcript current; only a + // watch-only session (which would never poll long) exits here. + if (!canSend) exit() + }) + .finally(() => { + streaming.current = false + }) + }, [canSend, exit, handleFrame, refreshSteps, sessionId, token, wsBase]) + + useEffect(() => { + pump() + scheduleRefresh() // catch steps created between the initial fetch and connect + // A slow poll backs up the socket wake (and covers a socket that never + // connected, so following still updates). The socket wake is the fast path; + // this is just a safety net, so it can be gentle. + const poll = setInterval(scheduleRefresh, 3000) + const controller = abort.current + return () => { + controller.abort() + clearInterval(poll) + if (refreshTimer.current) clearTimeout(refreshTimer.current) + } + }, [pump, scheduleRefresh]) + + // Tick an elapsed-seconds counter while the agent works, so a long turn (the + // stream delivers a whole message at once, not token-by-token) reads as live + // progress rather than a frozen spinner. Reset at the start of each turn. + useEffect(() => { + if (!working) return + setElapsed(0) + const t = setInterval(() => setElapsed((e) => e + 1), 1000) + return () => clearInterval(t) + }, [working]) + + // When the turn goes idle, commit any still-queued messages into the + // transcript (they weren't relayed) so they never vanish; remember them so a + // late backend echo is de-duplicated. A message sent while already idle + // flushes immediately — it starts a fresh turn rather than queueing. + useEffect(() => { + if (working || queued.length === 0) return + flushed.current.push(...queued) + const items = queued.map((text) => ({ + key: `q${keyCounter.current++}`, + kind: 'user', + gutter: '›', + text, + spaceBefore: true, + })) + setItems((prev) => [...prev, ...items]) + setQueued([]) + }, [working, queued]) + + const submit = useCallback( + (raw: string): void => { + const text = raw.trim() + setInput('') + if (!text) return + if (text === '/exit' || text === '/quit') { + exit() + return + } + void (async () => { + try { + if (text === '/stop') { + const s = await api.stopAgentSession(sessionId) + setNotice(`stop requested (${s.status}) — the conversation survives`) + return + } + // Show the message as queued (or, if idle, it flushes to the + // transcript at once), then post it and wait for the turn. + setQueued((prev) => [...prev, text]) + setNotice(null) + await api.sendSessionMessage(sessionId, text) + setWorking(true) + pump() + } catch (err) { + setQueued((prev) => { + const j = prev.indexOf(text) + return j < 0 ? prev : [...prev.slice(0, j), ...prev.slice(j + 1)] + }) + setNotice(`✗ ${err instanceof ApiError ? err.detail : (err as Error).message}`) + } + })() + }, + [api, exit, pump, sessionId], + ) + + const inputActive = canSend && isRawModeSupported + useInput( + (ch, key) => { + if (key.return) { + submit(input) + return + } + if (key.escape) { + if (working) submit('/stop') + return + } + if (key.ctrl && ch === 'r') { + setExpanded((v) => !v) + return + } + if (key.backspace || key.delete) { + setInput((p) => p.slice(0, -1)) + return + } + if (key.ctrl || key.meta) return + if (ch) setInput((p) => p + ch) + }, + { isActive: inputActive }, + ) + + // Render the transcript from state (not ) so ctrl+r can re-expand + // committed blocks. The list is memoized on [items, expanded], so the + // 80ms spinner ticks reuse the same elements and don't re-lay-out the + // transcript — only the footer re-renders. + const lines = useMemo( + () => items.map((item) => ), + [items, expanded], + ) + const hasCollapsible = useMemo( + () => items.some((it) => it.kind === 'tool_result' && it.text.split('\n').length > COLLAPSE_LINES), + [items], + ) + + return ( + + {lines} + + {notice && · {notice}} + {working && ( + + + + {' '} + + {status} · {elapsed}s{inputActive ? ' · esc to interrupt · type to queue a message' : ''} + + + )} + {inputActive ? ( + + › + {input} + + + ) : ( + !working && · {canSend ? status : `following ${status}`} + )} + {queued.length > 0 && ( + + {queued.map((text, i) => ( + + ⏳ queued · {text} + + ))} + + )} + {inputActive && hasCollapsible && ( + ctrl+r to {expanded ? 'collapse' : 'expand'} tool output + )} + + + ) +} + +// Tool-result bodies collapse to this many lines until ctrl+r expands them. +const COLLAPSE_LINES = 6 + +// Colour + weight for each transcript item kind, matched loosely to Claude Code. +function styleFor(item: TranscriptItem): { + gutterColor?: string + textColor?: string + dim: boolean + bold: boolean +} { + const kind: ItemKind = item.kind + switch (kind) { + case 'tool': + return { gutterColor: 'green', bold: true, dim: false } + case 'tool_result': + return { textColor: item.isError ? 'red' : undefined, dim: !item.isError, bold: false } + case 'user': + return { gutterColor: 'cyan', textColor: 'cyan', bold: true, dim: false } + case 'error': + return { gutterColor: 'red', textColor: 'red', dim: false, bold: false } + case 'summary': + return { textColor: item.isError ? 'red' : undefined, dim: true, bold: false } + case 'thinking': + case 'system': + case 'notice': + return { dim: true, bold: false } + case 'assistant': + default: + return { dim: false, bold: false } + } +} + +const TranscriptLine = React.memo(function TranscriptLine({ + item, + expanded, +}: { + item: TranscriptItem + expanded: boolean +}): React.ReactElement { + const mt = item.spaceBefore ? 1 : 0 + const { gutterColor, textColor, dim, bold } = styleFor(item) + + // Plain assistant prose: no gutter, just spaced text. + if (item.kind === 'assistant') { + return ( + + {item.text} + + ) + } + + // Tool results can be long; collapse to a compact body with a "+N lines" + // marker unless ctrl+r has expanded the transcript. + const clamped = + item.kind === 'tool_result' && !expanded + ? clampLines(item.text, COLLAPSE_LINES) + : { body: item.text, more: 0 } + + return ( + + + + {item.gutter ?? ''} + + + + + {clamped.body} + {item.detail ? {item.detail} : null} + + {clamped.more > 0 && … +{clamped.more} lines (ctrl+r to expand)} + + + ) +}) diff --git a/test/events.test.ts b/test/events.test.ts new file mode 100644 index 0000000..a0d2d08 --- /dev/null +++ b/test/events.test.ts @@ -0,0 +1,145 @@ +import { describe, expect, it } from 'vitest' +import { + clampLines, + eventToItems, + LineBuffer, + parseEventLine, + summarizeToolInput, + type CCEvent, +} from '../src/lib/events' + +describe('LineBuffer', () => { + it('emits only complete lines and buffers the trailing partial', () => { + const buf = new LineBuffer() + expect(buf.push('{"a":1}\n{"b":2}')).toEqual(['{"a":1}']) + expect(buf.push('\n{"c"')).toEqual(['{"b":2}']) + expect(buf.push(':3}\n')).toEqual(['{"c":3}']) + }) + + it('reassembles a single event split across frames', () => { + const buf = new LineBuffer() + expect(buf.push('{"ty')).toEqual([]) + expect(buf.push('pe":"x"}')).toEqual([]) + expect(buf.push('\n')).toEqual(['{"type":"x"}']) + }) + + it('flush releases a trailing partial once', () => { + const buf = new LineBuffer() + buf.push('tail') + expect(buf.flush()).toEqual(['tail']) + expect(buf.flush()).toEqual([]) + }) +}) + +describe('parseEventLine', () => { + it('parses a JSON object', () => { + expect(parseEventLine('{"type":"result"}')).toEqual({ type: 'result' }) + }) + + it('returns null for blank, non-JSON, or non-object lines', () => { + expect(parseEventLine(' ')).toBeNull() + expect(parseEventLine('not json')).toBeNull() + expect(parseEventLine('[1,2,3]')).toBeNull() + expect(parseEventLine('"a string"')).toBeNull() + }) +}) + +describe('summarizeToolInput', () => { + it('shows the file path for file tools', () => { + expect(summarizeToolInput('Read', { file_path: 'src/auth.ts' })).toBe('src/auth.ts') + expect(summarizeToolInput('Edit', { file_path: 'a.ts', old_string: 'x' })).toBe('a.ts') + }) + + it('shows the command for Bash and the pattern for Grep', () => { + expect(summarizeToolInput('Bash', { command: 'ls -la' })).toBe('ls -la') + expect(summarizeToolInput('Grep', { pattern: 'foo', path: 'src' })).toBe('foo in src') + }) + + it('falls back to compact JSON for unknown tools', () => { + expect(summarizeToolInput('Mystery', { a: 1 })).toBe('{"a":1}') + expect(summarizeToolInput('Mystery', {})).toBe('') + }) +}) + +describe('eventToItems', () => { + it('renders assistant text, thinking, and tool calls as grouped items', () => { + const event: CCEvent = { + type: 'assistant', + message: { + content: [ + { type: 'thinking', thinking: 'checking the auth flow' }, + { type: 'text', text: 'Reading the file.' }, + { type: 'tool_use', name: 'Read', input: { file_path: 'src/auth.ts' } }, + ], + }, + } + const items = eventToItems(event, 's1') + expect(items.map((i) => i.kind)).toEqual(['thinking', 'assistant', 'tool']) + expect(items[0].gutter).toBe('✻') + expect(items[2]).toMatchObject({ kind: 'tool', gutter: '●', text: 'Read', detail: '(src/auth.ts)' }) + // Unique keys within one event. + expect(new Set(items.map((i) => i.key)).size).toBe(items.length) + }) + + it('groups a tool_result under the call with a ⎿ gutter', () => { + const event: CCEvent = { + type: 'user', + message: { + content: [{ type: 'tool_result', tool_use_id: 't1', content: 'file contents' }], + }, + } + const [item] = eventToItems(event, 's2') + expect(item).toMatchObject({ kind: 'tool_result', gutter: '⎿', text: 'file contents', spaceBefore: false }) + }) + + it('unwraps nested text blocks in a tool_result and marks errors', () => { + const event: CCEvent = { + type: 'user', + message: { + content: [ + { type: 'tool_result', is_error: true, content: [{ type: 'text', text: 'boom' }] }, + ], + }, + } + const [item] = eventToItems(event, 's3') + expect(item.text).toBe('boom') + expect(item.isError).toBe(true) + }) + + it('renders a human user message with a › gutter (not as a tool result)', () => { + const event: CCEvent = { type: 'user', message: { content: 'please add tests' } } + const [item] = eventToItems(event, 's4') + expect(item).toMatchObject({ kind: 'user', gutter: '›', text: 'please add tests', spaceBefore: true }) + }) + + it('summarizes a result event with duration and cost', () => { + const event: CCEvent = { type: 'result', duration_ms: 12300, total_cost_usd: 0.042 } + const [item] = eventToItems(event, 's5') + expect(item.kind).toBe('summary') + expect(item.text).toBe('turn complete · 12.3s · $0.04') + }) + + it('renders a system init line', () => { + const event: CCEvent = { type: 'system', subtype: 'init', model: 'claude-opus-4-8' } + const [item] = eventToItems(event, 's6') + expect(item.kind).toBe('system') + expect(item.text).toContain('claude-opus-4-8') + }) + + it('produces no items for an empty assistant turn', () => { + expect(eventToItems({ type: 'assistant', message: { content: [] } }, 's7')).toEqual([]) + }) +}) + +describe('clampLines', () => { + it('leaves short bodies untouched', () => { + expect(clampLines('a\nb', 6)).toEqual({ body: 'a\nb', more: 0 }) + }) + + it('clamps long bodies and counts the remainder', () => { + const text = Array.from({ length: 10 }, (_, i) => `line ${i}`).join('\n') + const { body, more } = clampLines(text, 6) + expect(body.split('\n')).toHaveLength(6) + expect(more).toBe(4) + }) +})