From 0c451c1a764d275c4fd618de9339d67c9791e1ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Elberte=20Pl=C3=ADnio?= Date: Sun, 19 Jul 2026 10:34:07 -0300 Subject: [PATCH 1/2] feat(pi-kit): lane transcript capture and full-screen lanes TUI Raw per-lane JSONL transcripts under raw//.jsonl, a LaneTranscript reducer for live tailing and archived replay, a full-screen /lanes tui browser, and a logical line-count fix for compact read/write tool rows. --- bun.lock | 2 + packages/pi-kit/extensions/compact-tools.ts | 11 +- packages/pi-kit/extensions/lanes.ts | 22 +- packages/pi-kit/src/journal-core.ts | 17 +- packages/pi-kit/src/lanes-tui.ts | 353 ++++++++++++++++++++ packages/pi-kit/src/runner.ts | 21 ++ packages/pi-kit/src/transcript.ts | 113 +++++++ packages/pi-kit/test/transcript.test.ts | 49 +++ 8 files changed, 582 insertions(+), 6 deletions(-) create mode 100644 packages/pi-kit/src/lanes-tui.ts create mode 100644 packages/pi-kit/src/transcript.ts create mode 100644 packages/pi-kit/test/transcript.test.ts diff --git a/bun.lock b/bun.lock index 0747cee..37f06e6 100644 --- a/bun.lock +++ b/bun.lock @@ -44,7 +44,9 @@ "name": "@pickforge/pi-kit", "version": "0.1.0", "devDependencies": { + "@earendil-works/pi-ai": "0.79.10", "@earendil-works/pi-coding-agent": "0.79.10", + "@earendil-works/pi-tui": "0.79.10", "typebox": "1.1.38", "typescript": "^5.6.0", "vitest": "^2.0.0", diff --git a/packages/pi-kit/extensions/compact-tools.ts b/packages/pi-kit/extensions/compact-tools.ts index 3adef77..c3d7907 100644 --- a/packages/pi-kit/extensions/compact-tools.ts +++ b/packages/pi-kit/extensions/compact-tools.ts @@ -29,6 +29,13 @@ function firstLine(text: string): string { return idx === -1 ? text : text.slice(0, idx); } +/** Logical line count: a single trailing newline does not add a line. */ +function countLines(text: string): number { + if (text.length === 0) return 0; + const body = text.endsWith("\n") ? text.slice(0, -1) : text; + return body.split("\n").length; +} + function clip(text: string, width: number): string { return truncateToWidth(text, width, "…"); } @@ -102,7 +109,7 @@ export default function compactTools(pi: ExtensionAPI) { if (content?.type !== "text") return new Text(theme.fg("error", "✖ no content"), 0, 0); const details = result.details as ReadToolDetails | undefined; - const lineCount = content.text.split("\n").length; + const lineCount = countLines(content.text); let text = theme.fg("success", "✔ ") + theme.fg("muted", `${lineCount} lines`); if (details?.truncation?.truncated) { text += theme.fg("warning", ` of ${details.truncation.totalLines}`); @@ -173,7 +180,7 @@ export default function compactTools(pi: ExtensionAPI) { pi.registerTool({ ...write, renderCall(args, theme) { - const lineCount = args.content.split("\n").length; + const lineCount = countLines(args.content); return new Text( `${theme.fg("accent", "▸")} ${theme.fg("toolTitle", "write")} ${theme.fg("text", shortPath(args.path))} ${theme.fg("dim", `(${lineCount} lines)`)}`, 0, diff --git a/packages/pi-kit/extensions/lanes.ts b/packages/pi-kit/extensions/lanes.ts index 0b26ac8..071d5ff 100644 --- a/packages/pi-kit/extensions/lanes.ts +++ b/packages/pi-kit/extensions/lanes.ts @@ -1,6 +1,7 @@ import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; import { Type } from "typebox"; -import { appendEvent, newRunId } from "../src/journal-core.ts"; +import { appendEvent, newRunId, rawRunDir } from "../src/journal-core.ts"; +import { LanesTuiComponent } from "../src/lanes-tui.ts"; import { LaneRunner } from "../src/runner.ts"; import type { LaneProjection, LaneSpec, RunProjection } from "../src/schema.ts"; import { MODEL_TABLE, validateLaneSpec } from "../src/table.ts"; @@ -217,6 +218,7 @@ export default function lanesExtension(pi: ExtensionAPI): void { const activeRunner = new LaneRunner({ runId, append: appendEvent, + rawDir: rawRunDir(runId), onUpdate() { if (view && lastRun === view) renderWidget(ctx, view); }, @@ -427,9 +429,23 @@ export default function lanesExtension(pi: ExtensionAPI): void { }); pi.registerCommand("lanes", { - description: "Show, hide, summarize, or abandon lanes", + description: "Show, hide, summarize, abandon lanes, or open the full-screen TUI", handler: async (args, ctx) => { const [command = "show", lane] = (args.trim() || "show").split(/\s+/, 2); + if (command === "tui") { + if (!ctx.hasUI) { + ctx.ui.notify("lanes tui requires interactive mode", "warning"); + return; + } + await ctx.ui.custom( + (tui, theme, _keybindings, done) => new LanesTuiComponent(tui, theme, () => done(undefined), lastRun?.id), + { + overlay: true, + overlayOptions: { width: "100%", maxHeight: "100%", anchor: "top-left", row: 0, col: 0 }, + }, + ); + return; + } if (command === "hide") { if (lastRun) lastRun.hidden = true; ctx.ui.setWidget("pi-lanes", undefined); @@ -467,7 +483,7 @@ export default function lanesExtension(pi: ExtensionAPI): void { renderWidget(ctx, lastRun); return; } - ctx.ui.notify("Usage: /lanes [show|hide|last|abandon ]", "warning"); + ctx.ui.notify("Usage: /lanes [show|hide|last|tui|abandon ]", "warning"); }, }); } diff --git a/packages/pi-kit/src/journal-core.ts b/packages/pi-kit/src/journal-core.ts index 890594f..56b0b43 100644 --- a/packages/pi-kit/src/journal-core.ts +++ b/packages/pi-kit/src/journal-core.ts @@ -1,4 +1,4 @@ -import { appendFileSync, mkdirSync, readFileSync } from "node:fs"; +import { appendFileSync, mkdirSync, readdirSync, readFileSync } from "node:fs"; import { homedir } from "node:os"; import { join } from "node:path"; import { randomBytes } from "node:crypto"; @@ -19,6 +19,21 @@ export function journalDir(): string { return runsDir; } +/** Directory holding raw per-lane JSONL transcripts for a run. */ +export function rawRunDir(runId: string): string { + const dataDir = process.env[DATA_DIR_ENV] ?? join(homedir(), ".pickforge", "pi-kit"); + return join(dataDir, "raw", runId); +} + +/** All journaled run ids, newest first. */ +export function listRuns(): string[] { + return readdirSync(journalDir()) + .filter((name) => name.endsWith(".jsonl")) + .map((name) => name.slice(0, -".jsonl".length)) + .sort() + .reverse(); +} + export function appendEvent(ev: KitEvent): void { const event = { ...ev, diff --git a/packages/pi-kit/src/lanes-tui.ts b/packages/pi-kit/src/lanes-tui.ts new file mode 100644 index 0000000..32c3a52 --- /dev/null +++ b/packages/pi-kit/src/lanes-tui.ts @@ -0,0 +1,353 @@ +/** + * Full-screen lanes TUI overlay: browse live and archived lane runs, + * inspect each lane's session (thinking, text, tool calls, results). + * Shown via ctx.ui.custom() on top of the active Pi session. + */ +import { readFileSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { matchesKey, truncateToWidth, visibleWidth, wrapTextWithAnsi, type Component, type TUI } from "@earendil-works/pi-tui"; +import { listRuns, rawRunDir, readRun, reduceRun } from "./journal-core.ts"; +import type { LaneProjection, RunProjection } from "./schema.ts"; +import { LaneTranscript } from "./transcript.ts"; + +interface Theme { + fg(color: string, text: string): string; +} + +const RAIL_WIDTH = 30; +const POLL_MS = 500; + +interface LaneFeed { + transcript: LaneTranscript; + bytesRead: number; +} + +class RunView { + readonly runId: string; + projection: RunProjection; + private readonly feeds = new Map(); + + constructor(runId: string) { + this.runId = runId; + this.projection = reduceRun(readRun(runId)); + } + + refresh(): void { + this.projection = reduceRun(readRun(this.runId)); + for (const lane of this.projection.lanes.keys()) this.pollLane(lane); + } + + transcript(lane: string): LaneTranscript { + this.pollLane(lane); + return this.feeds.get(lane)!.transcript; + } + + private pollLane(lane: string): void { + let feed = this.feeds.get(lane); + if (!feed) { + feed = { transcript: new LaneTranscript(), bytesRead: 0 }; + this.feeds.set(lane, feed); + } + const path = join(rawRunDir(this.runId), `${lane}.jsonl`); + try { + const size = statSync(path).size; + if (size <= feed.bytesRead) return; + // Reread whole file and refeed only the new tail (lines are append-only). + const contents = readFileSync(path, "utf8"); + feed.transcript.feedChunk(contents.slice(feed.bytesRead)); + feed.bytesRead = size; + } catch { + // No transcript captured for this lane (pre-tap run or lane not started). + } + } +} + +function stateColor(state: LaneProjection["state"]): string { + switch (state) { + case "running": + return "accent"; + case "done": + return "success"; + case "failed": + case "abandoned": + return "error"; + default: + return "dim"; + } +} + +function stateGlyph(state: LaneProjection["state"]): string { + switch (state) { + case "running": + return "▶"; + case "done": + return "✔"; + case "failed": + return "✖"; + case "abandoned": + return "⊘"; + default: + return "◌"; + } +} + +export class LanesTuiComponent implements Component { + private runs: string[] = []; + private runIndex = 0; + private laneIndex = 0; + private pane: "runs" | "lanes" = "lanes"; + private scroll = 0; + private follow = true; + private showThinking = true; + private view?: RunView; + private timer?: NodeJS.Timeout; + private lineCache?: { key: string; lines: string[] }; + + private readonly tui: TUI; + private readonly theme: Theme; + private readonly done: () => void; + + constructor(tui: TUI, theme: Theme, done: () => void, initialRun?: string) { + this.tui = tui; + this.theme = theme; + this.done = done; + this.reloadRuns(initialRun); + this.timer = setInterval(() => { + this.view?.refresh(); + this.tui.requestRender(); + }, POLL_MS); + this.timer.unref(); + } + + dispose(): void { + clearInterval(this.timer); + } + + invalidate(): void { + this.lineCache = undefined; + } + + private reloadRuns(preferred?: string): void { + this.runs = listRuns(); + const index = preferred ? this.runs.indexOf(preferred) : 0; + this.runIndex = index >= 0 ? index : 0; + this.selectRun(); + } + + private selectRun(): void { + const runId = this.runs[this.runIndex]; + this.view = runId ? new RunView(runId) : undefined; + this.view?.refresh(); + this.laneIndex = 0; + this.scroll = 0; + this.follow = true; + this.lineCache = undefined; + } + + private lanes(): LaneProjection[] { + return this.view ? [...this.view.projection.lanes.values()] : []; + } + + private selectedLane(): LaneProjection | undefined { + return this.lanes()[this.laneIndex]; + } + + handleInput(data: string): void { + if (matchesKey(data, "q") || matchesKey(data, "escape")) { + this.done(); + return; + } + if (matchesKey(data, "tab")) { + this.pane = this.pane === "runs" ? "lanes" : "runs"; + return; + } + if (matchesKey(data, "up") || matchesKey(data, "k")) { + if (this.pane === "runs") { + this.runIndex = Math.max(0, this.runIndex - 1); + this.selectRun(); + } else { + this.laneIndex = Math.max(0, this.laneIndex - 1); + this.resetScroll(); + } + return; + } + if (matchesKey(data, "down") || matchesKey(data, "j")) { + if (this.pane === "runs") { + this.runIndex = Math.min(Math.max(0, this.runs.length - 1), this.runIndex + 1); + this.selectRun(); + } else { + this.laneIndex = Math.min(Math.max(0, this.lanes().length - 1), this.laneIndex + 1); + this.resetScroll(); + } + return; + } + if (matchesKey(data, "pageUp")) { + this.follow = false; + this.scroll = Math.max(0, this.scroll - this.viewHeight()); + return; + } + if (matchesKey(data, "pageDown")) { + this.scroll += this.viewHeight(); + return; + } + if (matchesKey(data, "t")) { + this.showThinking = !this.showThinking; + this.lineCache = undefined; + return; + } + if (matchesKey(data, "f")) { + this.follow = !this.follow; + return; + } + if (matchesKey(data, "r")) { + this.reloadRuns(this.runs[this.runIndex]); + } + } + + private resetScroll(): void { + this.scroll = 0; + this.follow = true; + this.lineCache = undefined; + } + + private viewHeight(): number { + return Math.max(4, this.tui.terminal.rows - 3); + } + + private transcriptLines(width: number): string[] { + const lane = this.selectedLane(); + if (!lane || !this.view) return [this.theme.fg("dim", "no lane selected")]; + const transcript = this.view.transcript(lane.spec.lane); + const key = `${this.view.runId}:${lane.spec.lane}:${transcript.version}:${width}:${this.showThinking}`; + if (this.lineCache?.key === key) return this.lineCache.lines; + + const t = this.theme; + const out: string[] = []; + const wrap = (text: string, color?: string) => { + for (const raw of text.split("\n")) { + const wrapped = wrapTextWithAnsi(raw, width); + for (const line of wrapped.length > 0 ? wrapped : [""]) { + out.push(color ? t.fg(color, line) : line); + } + } + }; + + if (transcript.entries.length === 0) { + out.push(t.fg("dim", "no transcript captured for this lane (pre-transcript run?)")); + out.push(""); + if (lane.answer) { + out.push(t.fg("accent", "── final answer (journal) ──")); + wrap(lane.answer); + } + } + for (const entry of transcript.entries) { + switch (entry.kind) { + case "task": + out.push(t.fg("accent", "── task ──")); + wrap(entry.text, "muted"); + break; + case "thinking": + if (!this.showThinking) break; + out.push(t.fg("dim", "── thinking ──")); + wrap(entry.text, "dim"); + break; + case "text": + out.push(""); + wrap(entry.text); + break; + case "tool": + out.push(t.fg("warning", `⚙ ${entry.title ?? "tool"}`)); + wrap(entry.text, "muted"); + break; + case "tool_result": { + const isError = (entry.title ?? "").endsWith("(error)"); + out.push(t.fg(isError ? "error" : "dim", `→ ${entry.title ?? "result"}`)); + wrap(entry.text.length > 1_500 ? `${entry.text.slice(0, 1_500)}…` : entry.text, "dim"); + break; + } + case "info": + wrap(entry.text, "dim"); + break; + } + } + this.lineCache = { key, lines: out }; + return out; + } + + private railLines(height: number): string[] { + const t = this.theme; + const width = RAIL_WIDTH - 2; + const lines: string[] = []; + const runsFocused = this.pane === "runs"; + + lines.push(t.fg(runsFocused ? "accent" : "muted", runsFocused ? "▸ RUNS (archived)" : " RUNS (archived)")); + const maxRuns = Math.max(3, Math.floor((height - 4) / 3)); + const runStart = Math.max(0, Math.min(this.runIndex - Math.floor(maxRuns / 2), this.runs.length - maxRuns)); + for (let i = runStart; i < Math.min(this.runs.length, runStart + maxRuns); i++) { + const selected = i === this.runIndex; + const label = truncateToWidth(this.runs[i]!.replace(/^run-/, ""), width - 2); + lines.push(selected ? t.fg(runsFocused ? "accent" : "text", `› ${label}`) : t.fg("dim", ` ${label}`)); + } + + lines.push(""); + const lanesFocused = this.pane === "lanes"; + lines.push(t.fg(lanesFocused ? "accent" : "muted", lanesFocused ? "▸ LANES" : " LANES")); + this.lanes().forEach((lane, i) => { + const selected = i === this.laneIndex; + const glyph = t.fg(stateColor(lane.state), stateGlyph(lane.state)); + const name = truncateToWidth(lane.spec.lane, width - 4); + lines.push(`${selected ? t.fg("accent", "›") : " "} ${glyph} ${selected ? t.fg("text", name) : t.fg("muted", name)}`); + const meta = truncateToWidth( + `${lane.spec.model.slice(lane.spec.model.indexOf("/") + 1)}:${lane.spec.effort} $${lane.cost.toFixed(3)}`, + width - 4, + ); + lines.push(` ${t.fg("dim", meta)}`); + }); + return lines.slice(0, height); + } + + render(width: number): string[] { + const height = Math.max(6, this.tui.terminal.rows); + const bodyHeight = height - 2; + const t = this.theme; + const view = this.view; + const lanes = this.lanes(); + const running = lanes.filter((l) => l.state === "running").length; + + const headerText = view + ? `lanes tui · ${view.runId} · ${lanes.length} lanes${running ? ` · ${running} running` : ""} · $${view.projection.totalCost.toFixed(4)}` + : "lanes tui · no runs recorded yet"; + const header = truncateToWidth(t.fg("accent", headerText), width); + + const railWidth = Math.min(RAIL_WIDTH, Math.max(16, Math.floor(width / 3))); + const mainWidth = Math.max(10, width - railWidth - 1); + const rail = this.railLines(bodyHeight); + const all = this.transcriptLines(mainWidth); + const viewH = bodyHeight; + const maxScroll = Math.max(0, all.length - viewH); + if (this.follow) this.scroll = maxScroll; + this.scroll = Math.min(this.scroll, maxScroll); + const visible = all.slice(this.scroll, this.scroll + viewH); + + const sep = t.fg("dim", "│"); + const lines: string[] = [header]; + for (let i = 0; i < bodyHeight; i++) { + const left = rail[i] ?? ""; + const leftPad = Math.max(0, railWidth - visibleWidth(left)); + const right = visible[i] ?? ""; + const row = `${truncateToWidth(left, railWidth)}${" ".repeat(leftPad)}${sep}${truncateToWidth(right, mainWidth)}`; + lines.push(truncateToWidth(row, width)); + } + + const pos = maxScroll > 0 ? ` · ${Math.min(100, Math.round((this.scroll / maxScroll) * 100))}%` : ""; + const footer = truncateToWidth( + t.fg( + "dim", + `tab panes · ↑↓ select · pgup/pgdn scroll${pos} · t thinking(${this.showThinking ? "on" : "off"}) · f follow(${this.follow ? "on" : "off"}) · r reload · q close`, + ), + width, + ); + lines.push(footer); + return lines; + } +} diff --git a/packages/pi-kit/src/runner.ts b/packages/pi-kit/src/runner.ts index 31e8612..f10c1a9 100644 --- a/packages/pi-kit/src/runner.ts +++ b/packages/pi-kit/src/runner.ts @@ -1,4 +1,6 @@ import { spawn, type ChildProcessByStdio } from "node:child_process"; +import { createWriteStream, mkdirSync, type WriteStream } from "node:fs"; +import { join } from "node:path"; import type { Readable } from "node:stream"; import type { KitEvent, LaneProjection, LaneSpec, RunProjection } from "./schema.ts"; import { estimateCost, validateLaneSpec } from "./table.ts"; @@ -9,10 +11,13 @@ export interface RunnerOptions { maxConcurrent?: number; piBinary?: string; onUpdate?: () => void; + /** Directory for raw per-lane JSONL transcripts (`/.jsonl`). */ + rawDir?: string; } interface LaneRuntime { child: ChildProcessByStdio; + raw?: WriteStream; finish: () => void; settled: boolean; startedAt: number; @@ -101,6 +106,7 @@ export class LaneRunner { private readonly maxConcurrent: number; private readonly onUpdate?: () => void; private readonly piBinary: string; + private readonly rawDir?: string; private readonly runId: string; private readonly runtimes = new Map(); private readonly live: RunProjection; @@ -111,6 +117,7 @@ export class LaneRunner { this.append = opts.append; this.maxConcurrent = Math.max(1, Math.floor(opts.maxConcurrent ?? 4)); this.piBinary = opts.piBinary ?? "pi"; + if (opts.rawDir) this.rawDir = opts.rawDir; this.onUpdate = opts.onUpdate; this.live = { run: opts.runId, @@ -174,6 +181,7 @@ export class LaneRunner { if (runtime) { runtime.settled = true; clearTimeout(runtime.statusTimer); + runtime.raw?.end(); this.terminate(runtime); } @@ -282,8 +290,19 @@ export class LaneRunner { }, ); + let raw: WriteStream | undefined; + if (this.rawDir) { + try { + mkdirSync(this.rawDir, { recursive: true }); + raw = createWriteStream(join(this.rawDir, `${spec.lane}.jsonl`), { flags: "a" }); + } catch { + // Transcript capture is best-effort; the lane must still run. + } + } + const runtime: LaneRuntime = { child, + ...(raw ? { raw } : {}), finish: resolve, settled: false, startedAt: Date.now(), @@ -304,6 +323,7 @@ export class LaneRunner { runtime.settled = true; clearTimeout(runtime.statusTimer); this.runtimes.delete(spec.lane); + runtime.raw?.end(); this.terminate(runtime); this.record({ v: 1, @@ -407,6 +427,7 @@ export class LaneRunner { const consumeLine = (line: string) => { if (!line.trim()) return; + if (runtime.raw?.writable) runtime.raw.write(`${line}\n`); try { handleEvent(JSON.parse(line)); } catch { diff --git a/packages/pi-kit/src/transcript.ts b/packages/pi-kit/src/transcript.ts new file mode 100644 index 0000000..af250b9 --- /dev/null +++ b/packages/pi-kit/src/transcript.ts @@ -0,0 +1,113 @@ +/** + * Lane transcript model: reduces a raw per-lane Pi JSONL stream (pi --mode json) + * into an ordered list of display entries (thinking, text, tool calls, results). + * Used by the lanes TUI for both live tailing and archived replay. + */ + +export type EntryKind = "task" | "thinking" | "text" | "tool" | "tool_result" | "info"; + +export interface TranscriptEntry { + kind: EntryKind; + /** Short label, e.g. tool name. */ + title?: string; + text: string; +} + +type JsonRecord = Record; + +function isRecord(value: unknown): value is JsonRecord { + return typeof value === "object" && value !== null; +} + +function firstText(content: unknown): string { + if (!Array.isArray(content)) return ""; + let text = ""; + for (const part of content) { + if (isRecord(part) && part.type === "text" && typeof part.text === "string") text += part.text; + } + return text; +} + +const MAX_ENTRY_TEXT = 20_000; +const MAX_RESULT_TEXT = 4_000; + +export class LaneTranscript { + readonly entries: TranscriptEntry[] = []; + /** Bumped on every visible change, so views can cheaply detect updates. */ + version = 0; + private currentKey = ""; + + /** Feed one raw JSONL line from the lane's pi process. */ + feed(line: string): void { + if (!line.trim()) return; + let event: unknown; + try { + event = JSON.parse(line); + } catch { + return; + } + if (!isRecord(event) || typeof event.type !== "string") return; + this.handle(event); + } + + feedChunk(chunk: string): void { + for (const line of chunk.split("\n")) this.feed(line); + } + + private push(kind: EntryKind, text: string, title?: string): void { + this.entries.push({ kind, text, ...(title !== undefined ? { title } : {}) }); + this.version++; + } + + private appendDelta(kind: "thinking" | "text", key: string, delta: string): void { + const last = this.entries[this.entries.length - 1]; + if (this.currentKey === key && last && last.kind === kind) { + if (last.text.length < MAX_ENTRY_TEXT) last.text += delta.slice(0, MAX_ENTRY_TEXT - last.text.length); + } else { + this.currentKey = key; + this.entries.push({ kind, text: delta.slice(0, MAX_ENTRY_TEXT) }); + } + this.version++; + } + + private handle(event: JsonRecord): void { + if (event.type === "message_start" && isRecord(event.message) && event.message.role === "user") { + if (this.entries.length === 0) this.push("task", firstText(event.message.content).slice(0, MAX_ENTRY_TEXT)); + return; + } + + if (event.type === "message_update" && isRecord(event.assistantMessageEvent)) { + const update = event.assistantMessageEvent; + const index = typeof update.contentIndex === "number" ? update.contentIndex : 0; + if (update.type === "thinking_delta" && typeof update.delta === "string") { + this.appendDelta("thinking", `thinking:${index}`, update.delta); + } else if (update.type === "text_delta" && typeof update.delta === "string") { + this.appendDelta("text", `text:${index}`, update.delta); + } else if (update.type === "toolcall_end" && isRecord(update.toolCall)) { + const call = update.toolCall; + const name = typeof call.name === "string" ? call.name : "tool"; + let args = ""; + try { + args = JSON.stringify(call.arguments) ?? ""; + } catch { + args = String(call.arguments); + } + this.currentKey = ""; + this.push("tool", args.slice(0, MAX_RESULT_TEXT), name); + } + return; + } + + if (event.type === "tool_execution_end") { + const name = typeof event.toolName === "string" ? event.toolName : "tool"; + const text = isRecord(event.result) ? firstText(event.result.content) : ""; + this.currentKey = ""; + this.push("tool_result", text.slice(0, MAX_RESULT_TEXT), event.isError === true ? `${name} (error)` : name); + return; + } + + if (event.type === "agent_end") { + this.currentKey = ""; + } + } +} diff --git a/packages/pi-kit/test/transcript.test.ts b/packages/pi-kit/test/transcript.test.ts new file mode 100644 index 0000000..0e5a4c5 --- /dev/null +++ b/packages/pi-kit/test/transcript.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from "vitest"; +import { LaneTranscript } from "../src/transcript.ts"; + +const line = (obj: unknown) => JSON.stringify(obj); + +describe("LaneTranscript", () => { + it("reduces a raw pi json stream into ordered entries", () => { + const t = new LaneTranscript(); + t.feed(line({ type: "message_start", message: { role: "user", content: [{ type: "text", text: "do the task" }] } })); + t.feed(line({ type: "message_update", assistantMessageEvent: { type: "thinking_delta", contentIndex: 0, delta: "hmm " } })); + t.feed(line({ type: "message_update", assistantMessageEvent: { type: "thinking_delta", contentIndex: 0, delta: "ok" } })); + t.feed(line({ type: "message_update", assistantMessageEvent: { type: "text_delta", contentIndex: 1, delta: "Answer: " } })); + t.feed(line({ type: "message_update", assistantMessageEvent: { type: "text_delta", contentIndex: 1, delta: "42" } })); + t.feed( + line({ + type: "message_update", + assistantMessageEvent: { + type: "toolcall_end", + contentIndex: 2, + toolCall: { name: "bash", arguments: { command: "echo hi" } }, + }, + }), + ); + t.feed( + line({ + type: "tool_execution_end", + toolName: "bash", + isError: false, + result: { content: [{ type: "text", text: "hi\n" }] }, + }), + ); + + expect(t.entries.map((entry) => entry.kind)).toEqual(["task", "thinking", "text", "tool", "tool_result"]); + expect(t.entries[0]!.text).toBe("do the task"); + expect(t.entries[1]!.text).toBe("hmm ok"); + expect(t.entries[2]!.text).toBe("Answer: 42"); + expect(t.entries[3]!.title).toBe("bash"); + expect(t.entries[4]!.text).toBe("hi\n"); + }); + + it("ignores malformed lines and keeps version monotonic", () => { + const t = new LaneTranscript(); + t.feedChunk("not json\n{}\n"); + expect(t.entries).toEqual([]); + const before = t.version; + t.feed(line({ type: "message_update", assistantMessageEvent: { type: "text_delta", contentIndex: 0, delta: "x" } })); + expect(t.version).toBeGreaterThan(before); + }); +}); From d88585b7d5e2e744ef9457de0eb826bbd3252221 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Elberte=20Pl=C3=ADnio?= Date: Sun, 19 Jul 2026 10:39:51 -0300 Subject: [PATCH 2/2] fix(pi-kit): byte-accurate transcript tailing and post-flush raw stream close Poll lane transcripts by byte offset with complete-line clamping so multibyte characters and half-flushed lines are never split or dropped; close the raw JSONL stream only after the child's final stdout flush so abandoned and failed lanes keep their transcript tail. --- packages/pi-kit/src/lanes-tui.ts | 22 +++++--- packages/pi-kit/src/runner.ts | 5 +- packages/pi-kit/test/lanes-tui.test.ts | 70 ++++++++++++++++++++++++++ 3 files changed, 89 insertions(+), 8 deletions(-) create mode 100644 packages/pi-kit/test/lanes-tui.test.ts diff --git a/packages/pi-kit/src/lanes-tui.ts b/packages/pi-kit/src/lanes-tui.ts index 32c3a52..c3271c3 100644 --- a/packages/pi-kit/src/lanes-tui.ts +++ b/packages/pi-kit/src/lanes-tui.ts @@ -3,7 +3,7 @@ * inspect each lane's session (thinking, text, tool calls, results). * Shown via ctx.ui.custom() on top of the active Pi session. */ -import { readFileSync, statSync } from "node:fs"; +import { closeSync, openSync, readSync, statSync } from "node:fs"; import { join } from "node:path"; import { matchesKey, truncateToWidth, visibleWidth, wrapTextWithAnsi, type Component, type TUI } from "@earendil-works/pi-tui"; import { listRuns, rawRunDir, readRun, reduceRun } from "./journal-core.ts"; @@ -22,7 +22,7 @@ interface LaneFeed { bytesRead: number; } -class RunView { +export class RunView { readonly runId: string; projection: RunProjection; private readonly feeds = new Map(); @@ -52,10 +52,20 @@ class RunView { try { const size = statSync(path).size; if (size <= feed.bytesRead) return; - // Reread whole file and refeed only the new tail (lines are append-only). - const contents = readFileSync(path, "utf8"); - feed.transcript.feedChunk(contents.slice(feed.bytesRead)); - feed.bytesRead = size; + // Track offsets in bytes (never UTF-16 code units) and feed only + // complete lines, so multibyte characters and half-flushed tail lines + // are never split or silently dropped. + const fd = openSync(path, "r"); + try { + const buf = Buffer.alloc(size - feed.bytesRead); + const read = readSync(fd, buf, 0, buf.length, feed.bytesRead); + const end = buf.subarray(0, read).lastIndexOf(0x0a) + 1; + if (end === 0) return; + feed.transcript.feedChunk(buf.subarray(0, end).toString("utf8")); + feed.bytesRead += end; + } finally { + closeSync(fd); + } } catch { // No transcript captured for this lane (pre-tap run or lane not started). } diff --git a/packages/pi-kit/src/runner.ts b/packages/pi-kit/src/runner.ts index f10c1a9..c60680e 100644 --- a/packages/pi-kit/src/runner.ts +++ b/packages/pi-kit/src/runner.ts @@ -181,7 +181,6 @@ export class LaneRunner { if (runtime) { runtime.settled = true; clearTimeout(runtime.statusTimer); - runtime.raw?.end(); this.terminate(runtime); } @@ -323,7 +322,6 @@ export class LaneRunner { runtime.settled = true; clearTimeout(runtime.statusTimer); this.runtimes.delete(spec.lane); - runtime.raw?.end(); this.terminate(runtime); this.record({ v: 1, @@ -465,6 +463,9 @@ export class LaneRunner { child.once("error", (error) => finish(false, runtime.stderrTail || error.message, `spawn:${error.message}`)); child.once("close", (code) => { if (runtime.stdoutBuffer) consumeLine(runtime.stdoutBuffer); + // End the transcript only after the last stdout flush so abandoned or + // failed lanes keep their tail lines. + runtime.raw?.end(); if (!runtime.settled) finish(false, runtime.stderrTail, `exit:${code ?? "signal"}`); }); return promise; diff --git a/packages/pi-kit/test/lanes-tui.test.ts b/packages/pi-kit/test/lanes-tui.test.ts new file mode 100644 index 0000000..71a118d --- /dev/null +++ b/packages/pi-kit/test/lanes-tui.test.ts @@ -0,0 +1,70 @@ +import { appendFileSync, mkdirSync, rmSync } from "node:fs"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { appendEvent, rawRunDir } from "../src/journal-core.ts"; +import { RunView } from "../src/lanes-tui.ts"; + +const originalDataDir = process.env.PIKIT_DATA_DIR; +let dataDir: string; + +const spec = { lane: "worker", task: "t", model: "openai-codex/gpt-5.6-sol", effort: "medium" as const }; + +function textEvent(text: string): string { + return JSON.stringify({ + type: "message_update", + assistantMessageEvent: { type: "text_delta", contentIndex: 0, delta: text }, + }); +} + +beforeEach(() => { + dataDir = mkdtempSync(join(tmpdir(), "pi-kit-tui-")); + process.env.PIKIT_DATA_DIR = dataDir; +}); + +afterEach(() => { + if (originalDataDir === undefined) delete process.env.PIKIT_DATA_DIR; + else process.env.PIKIT_DATA_DIR = originalDataDir; + rmSync(dataDir, { recursive: true, force: true }); +}); + +describe("RunView incremental transcript polling", () => { + it("tails multibyte content across polls without dropping events", () => { + const runId = "run-tui-test"; + appendEvent({ v: 1, t: new Date().toISOString(), run: runId, type: "run_created", lanes: 1, origin: "test" }); + appendEvent({ v: 1, t: new Date().toISOString(), run: runId, lane: "worker", type: "lane_created", spec }); + const dir = rawRunDir(runId); + mkdirSync(dir, { recursive: true }); + const file = join(dir, "worker.jsonl"); + + // First chunk: multibyte chars make byte length > UTF-16 length. + appendFileSync(file, `${textEvent("héllo — ✔ mündo")}\n`); + const view = new RunView(runId); + const first = view.transcript("worker"); + expect(first.entries.some((entry) => entry.text.includes("mündo"))).toBe(true); + + // Second chunk after the multibyte offset: must not be skipped or split. + appendFileSync(file, `${textEvent("segunda línea — completa")}\n`); + const second = view.transcript("worker"); + expect(second.entries.some((entry) => entry.text.includes("completa"))).toBe(true); + }); + + it("holds a partial trailing line until its newline arrives", () => { + const runId = "run-tui-partial"; + appendEvent({ v: 1, t: new Date().toISOString(), run: runId, type: "run_created", lanes: 1, origin: "test" }); + appendEvent({ v: 1, t: new Date().toISOString(), run: runId, lane: "worker", type: "lane_created", spec }); + const dir = rawRunDir(runId); + mkdirSync(dir, { recursive: true }); + const file = join(dir, "worker.jsonl"); + + const full = `${textEvent("only after newline")}\n`; + appendFileSync(file, full.slice(0, 25)); // half-flushed line + const view = new RunView(runId); + expect(view.transcript("worker").entries.length).toBe(0); + + appendFileSync(file, full.slice(25)); // rest arrives + const done = view.transcript("worker"); + expect(done.entries.some((entry) => entry.text.includes("only after newline"))).toBe(true); + }); +});