diff --git a/CHANGELOG.md b/CHANGELOG.md index e947372cf1..9613e1a5d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,10 @@ between releases do not reach them. Remote-PC runtime bundles are versioned sepa `grok-1.0.50-linux-x86_64` on Linux; other builds, macOS and Windows show the terminal view. See [setup and limits](docs/grok-chat.md). ([#777](https://github.com/devswha/herdr-web-ui/pull/777) by @infowolfe) +- **Experimental Antigravity chat** reads native messages and tool activity, follows the + running process’s active conversation, and honors saved rewind boundaries. macOS is untested. + See [compatibility and validation](docs/agy-chat.md). + ([#789](https://github.com/devswha/herdr-web-ui/pull/789) by @infowolfe) - Hermes's clarify questions get an answer card, as Claude Code's and Codex's do: one tap per choice, a typed answer for its "Other" row and for an open question, and ticks for a question that takes several answers. Text already on its answer line is cleared first, so it does not @@ -71,6 +75,9 @@ between releases do not reach them. Remote-PC runtime bundles are versioned sepa ([#673](https://github.com/devswha/herdr-web-ui/pull/673) by @WOULDU-pres) ### Fixed +- Chat hides a rejected conversation immediately, even when reading its terminal fallback + is slow or fails, so removed history does not remain on screen. + ([#789](https://github.com/devswha/herdr-web-ui/pull/789) by @infowolfe) - When a chat message is not sent because Claude Code's input box in the terminal is not empty, the note now says what the box holds, and a one-line draft there can be cleared from the chat with Clear it and send. The chat did not show that text, so Send seemed to do nothing until the diff --git a/docs/agy-chat.md b/docs/agy-chat.md new file mode 100644 index 0000000000..cd39a0a4d8 --- /dev/null +++ b/docs/agy-chat.md @@ -0,0 +1,54 @@ +# Antigravity Chat + +Chat reads the active Agy conversation's `transcript_full.jsonl`. User messages, +assistant text and expandable tool activity use the existing Chat view. Terminal +remains available for native commands and approvals. Reading Chat makes no model call. + +The reader follows the presence lock held by the pane's Agy process, including after +`/new` and `/resume`. It does not choose a session by directory, timestamp or startup +arguments, and needs no hook or status-line configuration. + +The ledger is replayed in order: repeated or backwards step indices replace the +corresponding suffix. A read-only query of the conversation database's step indices +also honors a saved rewind before replacement ledger records arrive. The reader does +not decode database payloads. Changed history invalidates older-page cursors and +expanded output references; ordinary appends preserve them. + +Planner calls and exported tool results have no shared native call ID in the tested +format, so results appear as independent tool-output rows. Hidden context and reasoning +are omitted. Image rendering and child-session expansion are outside this first scope. +The known `` wrapper is removed only when followed by the native +metadata suffix; standalone literal wrapper text is preserved. Error-only records +without text are not rendered. Only completed (`DONE`) exported steps are rendered; an unfinished last row waits, +and an unfinished historical row does not hide later completed turns. + +Linux process/lock discovery and native lifecycle behavior were tested with Agy 1.3.3. +macOS uses `lsof` for the same held-file evidence but has not been tested natively. +Its `ps` process-birth check has one-second resolution; same-second PID reuse is not distinguished. +Windows currently falls back to Terminal because this adapter has no native handle +lookup. A remote machine runs the reader on that machine, beside its agent store. + +Missing or ambiguous identity and unreadable stores use the existing terminal fallback. +An identified idle session with an empty database can show empty Chat before the ledger +exists. Until the ledger exists, discovery uses the monitor’s two-second retry; catalog +change notifications start after resolution. A new process without a conversation lock +still uses Terminal. A cold scan reads +up to 32 MiB per poll and resumes on the next poll; individual records are limited to +16 MiB and active history to 100,000 steps. Malformed complete step records use Terminal; +an incomplete final line waits for its newline. Rebuilding an evicted transcript index +invalidates its old cursors rather than reusing forgotten history state. + +The database and export are written separately. The reader checks active indices around +each read and remembers observed removals, but does not claim an atomic database/export +snapshot or detect a rewind-and-regrow completed entirely before its first observation. +Eviction or rebuilding of the bounded index also forgets observed removals: if the database +has already regained an index but its replacement ledger row has not arrived, old ledger +text can briefly reappear. A new history ID invalidates old cursors but does not remove +that cross-file lag window. No persistent retirement state or database payload decoding +is used. + +Validation covers fictional parser/paging fixtures, the conversation and whole-output +HTTP endpoints on an isolated Herdr server, and replay of an owned native conversation +through the production reader after saved rewind and restart. Native manual checks also +covered `/new`, `/resume`, repeated rewind, graceful restart and one crash while working. +These are bounded checks, not a guarantee about every Agy version or crash timing. diff --git a/docs/agy-chat.png b/docs/agy-chat.png new file mode 100644 index 0000000000..70ba09cada Binary files /dev/null and b/docs/agy-chat.png differ diff --git a/docs/guide.md b/docs/guide.md index 106106bd8b..0a044c9a20 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -153,6 +153,7 @@ Every agent herdr runs shows up with its live status, terminal and alerts. The c | **Devin CLI** | Native SQLite session, showing the active branch when herdr or the live process explicitly identifies the session; otherwise terminal text | — use Terminal | | **Grok** | Linux native update log with exact process/session evidence, on the `grok-1.0.50-linux-x86_64` build only; other builds use Terminal. See [Grok setup and limits](grok-chat.md). | — use Terminal | | **Hermes** | Native SQLite session store | — use Terminal | +| **Antigravity (agy)** | **Experimental; macOS untested.** Native ledger, selected by the running process’s held conversation lock. Linux native-tested; Windows uses terminal text. See [Agy behavior and limits](agy-chat.md). | — use Terminal | | **Anything else** | The terminal's text | — use Terminal | Devin sessions are never selected by directory or recency alone. A plain `devin` shows the diff --git a/scripts/chat-history-browser-qa.ts b/scripts/chat-history-browser-qa.ts index f142ee55ac..d224a32744 100644 --- a/scripts/chat-history-browser-qa.ts +++ b/scripts/chat-history-browser-qa.ts @@ -6,6 +6,7 @@ import { join } from "node:path"; import { chromium } from "playwright-core"; import { VOICE_DEFAULTS, type VoiceStatus } from "../shared/voice.ts"; import type {} from "./chat-history-fixture.tsx"; +import { parseAgyTranscript } from "../server/agy.ts"; const root = mkdtempSync(join(tmpdir(), "herdr-history-browser-")); let server: ReturnType | undefined; @@ -32,17 +33,32 @@ try { let epoch = "one", newest = "old newest", before: string | null = "one:100"; let holdOlder = false, olderRequested = false; let product = false; + let unavailable = false, terminalFails = false; + let holdTerminal = false, terminalReads = 0; + let terminalText = "Fictional terminal fallback"; /** product panes ("machine/pane") whose conversation requests wait for the script, and those that fail */ const held = new Set(), failing = new Set(); const pending: Array<{ key: string; answer: (status: number) => Promise }> = []; const user = (text: string) => ({ role: "user", ts: text, parts: [{ kind: "text", text }] }); + await page.route("**/api/**/read?*", async (route) => { + terminalReads += 1; + if (holdTerminal) await new Promise((resolve) => releases.push(resolve)); + await (terminalFails + ? route.fulfill({ status: 503, json: { error: { code: "timeout", message: "fictional terminal read failed" } } }) + : route.fulfill({ json: { read: { text: terminalText, truncated: false } } })); + }); await page.route("**/api/**/conversation?*", async (route) => { const url = new URL(route.request().url()); if (product) { const pane = url.searchParams.get("pane_id") ?? url.searchParams.get("pane") ?? ""; const key = `${/\/api\/machines\/([^/]+)\//.exec(url.pathname)?.[1] ?? "local"}/${pane}`; const answer = (status: number) => status === 200 - ? route.fulfill({ json: key === "local/devin" ? { source: "devin-transcript", history_id: "synthetic-devin-session", cursor: null, turns: [ + ? route.fulfill({ json: key === "local/agy" ? { source: "agy-transcript", history_id: "synthetic-agy-session", cursor: null, turns: parseAgyTranscript([ + { step_index: 0, type: "USER_INPUT", status: "DONE", content: "Summarize the last three fictional commits." }, + { step_index: 1, type: "PLANNER_RESPONSE", status: "DONE", tool_calls: [{ name: "run_command", args: { CommandLine: "git log -3 --oneline" } }] }, + { step_index: 2, type: "GENERIC", status: "DONE", content: "Fictional commit log" }, + { step_index: 3, type: "PLANNER_RESPONSE", status: "DONE", content: "Synthetic Agy answer" }, + ].map((row) => JSON.stringify(row)).join("\n")) } : key === "local/devin" ? { source: "devin-transcript", history_id: "synthetic-devin-session", cursor: null, turns: [ user("Synthetic Devin prompt"), { role: "assistant", ts: null, parts: [ { kind: "tool", name: "synthetic_tool", summary: "synthetic_tool", input: "{}", output: "synthetic result" }, @@ -56,6 +72,10 @@ try { else await answer(failing.has(key) ? 503 : 200); return; } + if (unavailable) { + await route.fulfill({ json: { source: "scrollback", turns: [] } }); + return; + } const captured = epoch; const cursor = url.searchParams.get("before") ?? url.searchParams.get("from"); if (cursor && !cursor.startsWith(`${epoch}:`)) { @@ -157,6 +177,58 @@ try { assert.deepEqual(errors, []); console.log("PASS reused tool ids after clear, machine switch and unmount cancellation; no browser errors"); + // An authoritative refusal must hide stale history even if the second, terminal read fails. + // This is the DB/export mismatch path: the reader returns scrollback, not a transport error. + for (const terminalFailure of [false, true]) { + epoch = `refusal-${terminalFailure}`; newest = `discarded newest ${terminalFailure}`; before = `${epoch}:100`; + await refresh(); + await page.getByText(newest, { exact: true }).waitFor(); + await page.locator(".chat-older").click(); + await page.getByText("old earlier", { exact: true }).waitFor(); + await page.locator(".work-block-head").click(); + await page.locator(".work-row-head").click(); + await page.locator(".chat-tool-more").click(); + const lateOutput = await page.evaluate(() => window.qa.requests.length - 1); + const readsBefore = terminalReads; + terminalFails = terminalFailure; unavailable = true; holdTerminal = true; + await refresh(); + for (const deadline = Date.now() + 10_000; terminalReads === readsBefore;) { + assert.ok(Date.now() < deadline, "missing terminal fallback request"); + await Bun.sleep(20); + } + await page.evaluate(() => new Promise((resolve) => requestAnimationFrame(() => requestAnimationFrame(() => resolve())))); + assert.equal(await page.getByText(newest, { exact: true }).count(), 0, "refused history must disappear while terminal fallback is still pending"); + holdTerminal = false; + releases.splice(0).forEach((release) => release()); + if (terminalFailure) await page.locator(".chat-inline-error").filter({ hasText: "fictional terminal read failed" }).waitFor(); + else await page.getByText("Fictional terminal fallback", { exact: true }).waitFor(); + assert.equal(await page.getByText(newest, { exact: true }).count(), 0, "refused latest history must disappear even when terminal fallback fails"); + assert.equal(await page.getByText("old earlier", { exact: true }).count(), 0, "refused older pages must disappear"); + await page.waitForFunction((i) => window.qa.requests[i].signal.aborted, lateOutput); + await page.evaluate((i) => window.qa.requests[i].resolve("REFUSED TOOL OUTPUT"), lateOutput); + await page.evaluate(() => new Promise((resolve) => requestAnimationFrame(() => requestAnimationFrame(() => resolve())))); + assert.equal(await page.getByText("REFUSED TOOL OUTPUT", { exact: true }).count(), 0); + unavailable = false; terminalFails = false; before = null; + await refresh(); + await page.getByText(newest, { exact: true }).waitFor(); + assert.equal(await page.getByText("old earlier", { exact: true }).count(), 0); + } + console.log("PASS refusal clears latest/older history and late whole output, including failed terminal fallback; recovery reloads"); + + // Keep the existing scrollback change detector: clearing the signature on every fallback + // poll would suppress the New messages chip while the reader is scrolled up. + unavailable = true; + terminalText = Array.from({ length: 100 }, (_, i) => `Fictional terminal line ${i}`).join("\n"); + await refresh(); + await page.getByText(/Fictional terminal line 99/).waitFor(); + await page.getByRole("log").evaluate((node) => { node.scrollTop = 0; node.dispatchEvent(new Event("scroll")); }); + terminalText += "\nFictional terminal appended line"; + await refresh(); + await page.getByText(/Fictional terminal appended line/).waitFor(); + await page.locator(".chat-new-messages").waitFor(); + unavailable = false; + console.log("PASS changing terminal fallback retains the New messages indicator while scrolled up"); + // Exercise the actual product owner. A profiler observes every committed DOM, not just the final frame. const select = async (pane: string, machine: string, agent?: string) => { product = true; @@ -223,6 +295,15 @@ try { assert.deepEqual(await activity.json(), { tasks: [], runs: [] }); assert.equal(await page.getByText("Synthetic Devin answer", { exact: true }).count(), 0); assert.ok((await page.evaluate(() => window.qa.commits)).every((turns) => !turns.some((turn) => turn.includes("Synthetic Devin"))), "switch to Claude cannot commit Devin history"); + await select("agy", "local", "agy"); + await page.getByText("Summarize the last three fictional commits.", { exact: true }).waitFor(); + await page.getByText("Synthetic Agy answer", { exact: true }).waitFor(); + await page.locator(".work-block-head").click(); + assert.equal(await page.locator(".work-row-head").count(), 2, "Agy planner call and independent result remain separate"); + await page.locator(".work-row-head").last().click(); + await page.getByText("Fictional commit log", { exact: true }).waitFor(); + assert.equal(await page.getByText("Synthetic Claude prompt", { exact: true }).count(), 0); + console.log("PASS Agy native parser output renders user/assistant text and independent expandable tool results"); assert.deepEqual(errors, []); console.log("PASS synthetic Devin bubbles and tools render; switching to Claude clears Devin history"); console.log("PASS product pane commits across A→B→A, another PC, and late or failed answers"); diff --git a/server/agy-store.ts b/server/agy-store.ts new file mode 100644 index 0000000000..205689f005 --- /dev/null +++ b/server/agy-store.ts @@ -0,0 +1,87 @@ +/** Agy owns one presence lock for the conversation currently displayed in this process. */ +import { readFileSync, opendirSync, readlinkSync, realpathSync, statSync } from "node:fs"; +import { basename, dirname, join } from "node:path"; +import type { HerdrPane } from "../shared/protocol.ts"; +import { herdrRpc } from "./herdr/client.ts"; + +export interface AgyStore { root: string; session: string; path: string; pid: number; started: string } +const LOCK = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.lock$/i; + +/** Only paths held by the selected process are candidates; directory contents aren't evidence. */ +export function agyStoreFromLocks(paths: readonly string[]): Pick | null { + const locks = new Set(); + for (const path of paths) { + if (basename(dirname(path)) !== "presence" || !LOCK.test(basename(path))) continue; + try { + const canonical = realpathSync(path); + if (basename(dirname(canonical)) !== "presence" || !LOCK.test(basename(canonical)) || !statSync(canonical).isFile()) return null; + locks.add(canonical); + } catch { return null; } + } + if (locks.size !== 1) return null; + const lock = [...locks][0]!; + try { + const root = dirname(dirname(lock)); + const session = basename(lock, ".lock"); + const path = join(root, "brain", session, ".system_generated", "logs", "transcript_full.jsonl"); + return { root, session, path }; + } catch { return null; } +} + +/** Stable process birth identity; elapsed-time estimates drift between reads on macOS. */ +async function processIdentity(pid: number): Promise<{ started: string } | null> { + try { + if (process.platform === "linux") { + const fields = readFileSync(`/proc/${pid}/stat`, "utf8").split(") ").pop()!.split(" "); + const ticks = fields[19]; + if (!ticks || !/^\d+$/.test(ticks)) return null; + return { started: `${readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim()}:${ticks}` }; + } + if (process.platform === "darwin") { + const child = Bun.spawn(["/bin/ps", "-o", "lstart=", "-p", String(pid)], { + stdout: "pipe", stderr: "ignore", timeout: 1000, env: { ...process.env, LC_ALL: "C", TZ: "UTC" }, + }); + const started = (await new Response(child.stdout).text()).trim(); + return await child.exited === 0 && started ? { started } : null; + } + } catch { /* gone or unreadable owner */ } + return null; +} + +async function heldPaths(pid: number): Promise { + if (process.platform === "darwin") { + const child = Bun.spawn(["/usr/sbin/lsof", "-nPbw", "-a", "-p", String(pid), "-Fn"], { stdout: "pipe", stderr: "ignore", timeout: 3000 }); + const text = await new Response(child.stdout).text(); + if (await child.exited !== 0 || text.length > 1024 * 1024) return []; + return text.split("\n").filter((line) => line.startsWith("n")).map((line) => line.slice(1)); + } + if (process.platform !== "linux") return []; + const dir = opendirSync(`/proc/${pid}/fd`); + const paths: string[] = []; + try { + let count = 0; + for (let entry = dir.readSync(); entry; entry = dir.readSync()) { + if (++count > 4096) return []; + try { paths.push(readlinkSync(`/proc/${pid}/fd/${entry.name}`)); } catch { /* closed during enumeration */ } + } + } finally { dir.closeSync(); } + return paths; +} + +export async function resolveAgyStore(pane: HerdrPane): Promise { + if (pane.agent !== "agy") return null; + try { + const info = await herdrRpc<{ process_info?: { foreground_processes?: { pid: number; name?: string; argv?: string[] }[] } }>("pane.process_info", { pane_id: pane.pane_id }); + const owners = (info.process_info?.foreground_processes ?? []).filter((p) => p.name === "agy" && Number.isSafeInteger(p.pid) && p.pid > 1); + if (owners.length !== 1) return null; + const pid = owners[0]!.pid; + const before = await processIdentity(pid); + const store = agyStoreFromLocks(await heldPaths(pid)); + const after = await processIdentity(pid); + return before && store && before.started === after?.started ? { ...store, pid, started: before.started } : null; + } catch { return null; } +} + +export function sameAgyStore(a: AgyStore, b: AgyStore | null): boolean { + return b !== null && a.pid === b.pid && a.started === b.started && a.path === b.path; +} diff --git a/server/agy.ts b/server/agy.ts new file mode 100644 index 0000000000..77e70c85f3 --- /dev/null +++ b/server/agy.ts @@ -0,0 +1,202 @@ +/** Agy's append-only step ledger, reduced in physical order before shared paging. */ +import { Database } from "bun:sqlite"; +import { createHash, randomUUID } from "node:crypto"; +import { closeSync, fstatSync, openSync, readSync, realpathSync, statSync } from "node:fs"; +import { basename, dirname, join } from "node:path"; +import type { ConversationPart, ConversationTurn } from "../shared/protocol.ts"; +import { trimOutput } from "./tool-output.ts"; +import { toolSummary } from "./transcript-records.ts"; + +const MAX_ENTRIES = 100_000; +const MAX_LINE = 16 * 1024 * 1024; +const SCAN_BYTES = 32 * 1024 * 1024; +const hash = (value: string): string => createHash("sha256").update(value).digest("base64url"); +const record = (value: unknown): Record => value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +export class AgyUnavailable extends Error { constructor() { super("Agy history is unavailable"); } } +interface Step { index: number; start: number; end: number; type: string; status: string; digest: string } +interface Index { identity: string; size: number; changed: string; scanned: number; revision: string; steps: Step[]; visible: Step[]; retired: Set } +const indexes = new Map(); +export function forgetAgyState(path?: string): void { if (path === undefined) indexes.clear(); else indexes.delete(path); } + +/** The catalog beside this resolved native ledger; also watched for saved rewinds. */ +export function agyDatabasePath(path: string): string { + const sessionDir = dirname(dirname(dirname(path))); + return join(dirname(dirname(sessionDir)), "conversations", `${basename(sessionDir)}.db`); +} + +/** The database supplies only the active indices, never payloads or guessed numeric statuses. */ +export function agyCatalog(path: string): Set { + const dbPath = agyDatabasePath(path); + let db: Database | undefined; + try { + if (realpathSync(dbPath) !== dbPath) throw new AgyUnavailable(); + db = new Database(dbPath, { readonly: true }); + db.exec("PRAGMA busy_timeout = 100"); + const rows = db.query<{ idx: number }, []>(`SELECT idx FROM steps ORDER BY idx LIMIT ${MAX_ENTRIES + 1}`).all(); + if (rows.length > MAX_ENTRIES || rows.some((row) => !Number.isSafeInteger(row.idx) || row.idx < 0)) throw new AgyUnavailable(); + return new Set(rows.map((row) => row.idx)); + } catch { throw new AgyUnavailable(); } + finally { db?.close(); } +} + +function entry(line: string): Record { + let value: Record; + try { value = record(JSON.parse(line)); } catch { throw new AgyUnavailable(); } + if (!Number.isSafeInteger(value.step_index) || (value.step_index as number) < 0 || typeof value.type !== "string") throw new AgyUnavailable(); + return value; +} + +export interface AgyProjection { segments: { start: number; end: number }[]; promptStarts: number[]; revision: string; indices: number[] } +export function agyProjection(path: string, fd: number): AgyProjection { + const stat = fstatSync(fd); + const identity = `${stat.dev}:${stat.ino}`; + const changed = `${stat.mtimeMs}:${stat.ctimeMs}`; + let index = indexes.get(path); + if (!index || index.identity !== identity || stat.size < index.size || (stat.size === index.size && changed !== index.changed)) { + index = { identity, size: stat.size, changed, scanned: 0, revision: randomUUID(), steps: [], visible: [], retired: new Set() }; + } + index.size = stat.size; index.changed = changed; + indexes.delete(path); indexes.set(path, index); + if (indexes.size > 16) indexes.delete(indexes.keys().next().value!); + let position = index.scanned; + let carry: Buffer[] = []; + let carried = 0; + const end = Math.min(stat.size, position + SCAN_BYTES); + while (position < end) { + const chunk = Buffer.alloc(Math.min(64 * 1024, end - position)); + const read = readSync(fd, chunk, 0, chunk.length, position); + if (!read) throw new AgyUnavailable(); + const bytes = chunk.subarray(0, read); + let start = 0; + for (let newline = bytes.indexOf(10); newline !== -1; newline = bytes.indexOf(10, start)) { + if (carried + newline - start > MAX_LINE) throw new AgyUnavailable(); + const line = (carried ? Buffer.concat([...carry, bytes.subarray(start, newline)]) : bytes.subarray(start, newline)).toString("utf8"); + const lineStart = position + start - carried; + const lineEnd = position + newline + 1; + carry = []; carried = 0; + const value = entry(line); + const number = value.step_index as number; + if (index.steps.length && number <= index.steps.at(-1)!.index) { + while (index.steps.length && index.steps.at(-1)!.index >= number) index.steps.pop(); + } + index.steps.push({ index: number, start: lineStart, end: lineEnd, type: value.type as string, status: String(value.status ?? ""), digest: hash(line) }); + if (index.steps.length > MAX_ENTRIES) throw new AgyUnavailable(); + index.scanned = lineEnd; + start = newline + 1; + } + if (start < bytes.length) { carry.push(bytes.subarray(start)); carried += bytes.length - start; } + if (carried > MAX_LINE) throw new AgyUnavailable(); + position += read; + } + if (position < stat.size) throw new AgyUnavailable(); + const catalog = agyCatalog(path); + // Native idle rewind updates the DB before any replacement ledger lines arrive. + // DB-only steps are normal; equality or a strict prefix would reject healthy logs. + const physicalStarts = new Set(index.steps.map((step) => step.start)); + for (const start of index.retired) if (!physicalStarts.has(start)) index.retired.delete(start); + // Once the active DB removed a displayed step, its old bytes cannot return merely + // because the replacement DB row precedes the replacement export. Never-published + // lagging steps remain eligible. The set is bounded by the active physical index. + for (const step of index.visible) if (!catalog.has(step.index) && physicalStarts.has(step.start)) index.retired.add(step.start); + const retained = index.steps.filter((step) => catalog.has(step.index) && !index.retired.has(step.start)); + const visible = retained.at(-1)?.status !== "DONE" ? retained.slice(0, -1) : retained; + // A DB snapshot can lag an exported append. Keep the physical index intact so the + // next poll can admit it. Only changes to already published spans invalidate paging. + if (index.visible.some((step, at) => visible[at]?.start !== step.start || visible[at]?.end !== step.end)) index.revision = randomUUID(); + index.visible = visible; + const segments: AgyProjection["segments"] = []; + const promptStarts: number[] = [], indices: number[] = []; + let length = 0; + for (const step of visible) { + indices.push(step.index); + if (step.type === "USER_INPUT" && step.status === "DONE") promptStarts.push(length); + const previous = segments.at(-1); + if (previous?.end === step.start) previous.end = step.end; + else segments.push({ start: step.start, end: step.end }); + length += step.end - step.start; + } + return { segments, promptStarts, revision: index.revision, indices }; +} + +/** Keep one fd through projection and page reads, and reject a cut during the read. */ +export function withAgyHistory(path: string, read: (projection: AgyProjection, fd: number) => T): T { + const fd = openSync(path, "r"); + try { + if (realpathSync(path) !== path) throw new AgyUnavailable(); + const before = fstatSync(fd); + const projection = agyProjection(path, fd); + const result = read(projection, fd); + const after = statSync(path), pinned = fstatSync(fd); + const current = agyCatalog(path); + if (before.dev !== after.dev || before.ino !== after.ino || pinned.size < before.size + || (pinned.size === before.size && (pinned.mtimeMs !== before.mtimeMs || pinned.ctimeMs !== before.ctimeMs)) + || projection.indices.some((index) => !current.has(index))) throw new AgyUnavailable(); + return result; + } finally { closeSync(fd); } +} + +function userText(text: string): string { + if (!text.startsWith("")) return text; + const end = text.lastIndexOf(""); + if (end < 0) return text; + const suffix = text.slice(end + "".length); + if (!/^\s*<(?:ADDITIONAL_METADATA|USER_SETTINGS_CHANGE)>/.test(suffix)) return text; + return text.slice("".length, end).replace(/^\n|\n$/g, ""); +} +function outputRef(path: string, step: Step): string { + const index = indexes.get(path)!; + return `agy:${hash(JSON.stringify([path, index.identity, index.revision, step.start, step.digest]))}`; +} +export function parseAgyTranscript(text: string, path?: string): ConversationTurn[] { + const turns: ConversationTurn[] = []; + const steps = path ? new Map(indexes.get(path)?.visible.map((step) => [step.index, step])) : undefined; + for (const line of text.split("\n")) { + if (!line.trim()) continue; + const value = entry(line); + if (value.status !== "DONE") continue; + const content = typeof value.content === "string" ? value.content : ""; + const ts = typeof value.created_at === "string" ? value.created_at : null; + if (value.type === "USER_INPUT") { turns.push({ role: "user", ts, parts: [{ kind: "text", text: userText(content) }] }); continue; } + if (value.type !== "PLANNER_RESPONSE" && value.type !== "GENERIC") continue; + const parts: ConversationPart[] = []; + if (value.type === "GENERIC") { + if (!content) continue; + // Native exports provide no call ID. Keep this result independent of planner calls. + const part: Extract = { kind: "tool", name: "Tool output", summary: `Step ${value.step_index}`, input: "", output: "" }; + const step = steps?.get(value.step_index as number); + trimOutput(part, content, path && step ? outputRef(path, step) : `agy:${hash(line)}`); + parts.push(part); + } else { + if (content) parts.push({ kind: "text", text: content }); + for (const call of Array.isArray(value.tool_calls) ? value.tool_calls : []) { + const tool = record(call), input = record(tool.args); + if (typeof tool.name !== "string") continue; + const mapped = { ...input, command: input.CommandLine, file_path: input.AbsolutePath ?? input.TargetFile }; + parts.push({ kind: "tool", name: tool.name, summary: toolSummary(tool.name, mapped), input: JSON.stringify(input), output: "" }); + } + } + if (!parts.length) continue; + const previous = turns.at(-1); + if (previous?.role === "assistant") { previous.parts.push(...parts); previous.end_ts = ts ?? undefined; } + else turns.push({ role: "assistant", ts, end_ts: ts ?? undefined, parts }); + } + return turns; +} + +export function agyToolOutput(path: string, ref: string): string | null { + if (!/^agy:[A-Za-z0-9_-]{43}$/.test(ref)) return null; + try { + return withAgyHistory(path, (projection, fd) => { + const visible = new Set(projection.indices); + for (const step of indexes.get(path)!.visible) { + if (step.type !== "GENERIC" || !visible.has(step.index) || outputRef(path, step) !== ref) continue; + const bytes = Buffer.alloc(step.end - step.start); + if (readSync(fd, bytes, 0, bytes.length, step.start) !== bytes.length) return null; + const value = entry(bytes.toString("utf8")); + const text = typeof value.content === "string" ? value.content : ""; + return text.length > 2_000_000 ? `${text.slice(0, 2_000_000)}\n… trimmed` : text; + } + return null; + }); + } catch { return null; } +} diff --git a/server/api.contract.test.ts b/server/api.contract.test.ts index 6a0fa1b120..789cc9a9ce 100644 --- a/server/api.contract.test.ts +++ b/server/api.contract.test.ts @@ -3,6 +3,7 @@ import { Database } from "bun:sqlite"; import { appendFileSync, chmodSync, copyFileSync, mkdirSync, mkdtempSync, existsSync, readFileSync, realpathSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { basename, join } from "node:path"; +import { ConversationMonitor } from "./conversation-monitor.ts"; import { createServer } from "./index.ts"; import type { AgentKind, AgentStatus, ApiError, HealthAuth, IntegrationsResponse, PushKey, RemoteAccess, SessionSnapshot, TabMoved, PaneFindResponse, PaneReadResult, UsageReport, WorkspaceCreated, WorktreeListing, WorktreeOpened, WorktreeRemoved } from "../shared/protocol.ts"; import { HerdrUpdater } from "./herdr-update.ts"; @@ -110,6 +111,75 @@ describe("Grok conversation API", () => { }, 20_000); }); +describe("Agy conversation API", () => { + it.skipIf(process.platform !== "linux")("binds the selected held lock, honors saved rewinds and recovers from an unwritten store", async () => { + const root = mkdtempSync(join(tmpdir(), "herdr-web-ui-agy-contract-")); + const id = "00000000-0000-4000-8000-000000000031"; + const path = join(root, "brain", id, ".system_generated", "logs", "transcript_full.jsonl"); + mkdirSync(join(path, ".."), { recursive: true }); + mkdirSync(join(root, "conversations")); mkdirSync(join(root, "presence")); + const lock = join(root, "presence", `${id}.lock`); + writeFileSync(lock, ""); + // Another unheld lock in the same store must never select a conversation. + writeFileSync(join(root, "presence", "00000000-0000-4000-8000-000000000032.lock"), ""); + const executable = join(root, "agy"); + copyFileSync("/bin/sleep", executable); chmodSync(executable, 0o755); + const created = await workspaceCreate({ cwd: root, label: "herdr-web-ui-test-agy-api" }); + let app: ReturnType | undefined, db: Database | undefined; + let monitor: ConversationMonitor | undefined; + try { + const paneId = created.root_pane.pane_id; + await herdrRpc("pane.send_text", { pane_id: paneId, text: `${executable} 600 3<${lock}\n` }); + for (const deadline = Date.now() + 10_000;;) { + if ((await sessionSnapshot()).panes.find((pane) => pane.pane_id === paneId)?.agent === "agy") break; + if (Date.now() > deadline) throw new Error("fixture Agy process did not start"); + await Bun.sleep(50); + } + app = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "state") }); + const url = `http://127.0.0.1:${app.port}/api/pane/conversation?pane_id=${encodeURIComponent(paneId)}`; + expect(await (await fetch(url)).json()).toEqual({ source: "scrollback", turns: [] }); + db = new Database(join(root, "conversations", `${id}.db`)); + db.exec("PRAGMA journal_mode=WAL; CREATE TABLE steps(idx INTEGER PRIMARY KEY)"); + const empty = await (await fetch(url)).json(); + expect(empty.source).toBe("agy-transcript"); expect(empty.turns).toEqual([]); + const row = (step_index: number, type: string, content: string) => JSON.stringify({ step_index, type, status: "DONE", content }) + "\n"; + db.exec("INSERT INTO steps VALUES (0), (1)"); + writeFileSync(path, row(0, "USER_INPUT", "fictional prompt") + row(1, "GENERIC", "x".repeat(5000))); + const first = await fetch(url), body = await first.json(); + expect(body.source).toBe("agy-transcript"); + expect(body.turns[0].parts[0].text).toBe("fictional prompt"); + expect((await fetch(url, { headers: { "if-none-match": first.headers.get("etag")! } })).status).toBe(304); + const ref = body.turns[1].parts[0].output_ref; + const outputUrl = `http://127.0.0.1:${app.port}/api/pane/conversation/tool-output?${new URLSearchParams({ pane_id: paneId, ref })}`; + expect(await (await fetch(outputUrl)).text()).toBe("x".repeat(5000)); + const changes: string[] = []; + monitor = new ConversationMonitor((_pane, signature) => changes.push(signature)); + monitor.watch(paneId); + const changed = async (count: number) => { + for (const deadline = Date.now() + 5000; changes.length <= count;) { + if (Date.now() > deadline) throw new Error("Agy catalog change was not pushed"); + await Bun.sleep(20); + } + }; + await changed(0); + const beforeCut = changes.length; + db.exec("DELETE FROM steps WHERE idx >= 1"); + await changed(beforeCut); + expect((await (await fetch(url)).json()).turns).toHaveLength(1); + expect((await fetch(`${url}&from=${encodeURIComponent(`${body.history_id}:0`)}`)).status).toBe(409); + expect((await fetch(outputUrl)).status).toBe(404); + forgetTranscriptState(); + const local = await fetch(url.replace("/api/pane/", "/api/machines/local/pane/")); + expect((await local.json()).turns).toHaveLength(1); + db.exec("DROP TABLE steps"); + expect(await (await fetch(url)).json()).toEqual({ source: "scrollback", turns: [] }); + } finally { + monitor?.stop(); app?.stop(); db?.close(); await workspaceClose(created.workspace.workspace_id); + forgetTranscriptState(); rmSync(root, { recursive: true, force: true }); + } + }, 20_000); +}); + describe("single-pane context lookup", () => { for (const route of ["commands", "files"] as const) { it(`rejects an unknown pane on /api/pane/${route} with the shared error envelope`, async () => { diff --git a/server/conversation-metadata.ts b/server/conversation-metadata.ts index 9a2b5ccb52..37df675639 100644 --- a/server/conversation-metadata.ts +++ b/server/conversation-metadata.ts @@ -36,6 +36,8 @@ function contextOf(used: number, model: string | null, window: number | null, be export function parseConversationMetadata(text: string, source: ConversationResponse["source"], before?: ConversationMetadata): ConversationMetadata { // `before` is what the text preceding this one said: the fold picks up where it stopped const metadata: ConversationMetadata = before ? { ...before } : { model: null, reasoning_effort: null }; + // Agy's exported rows do not attest the selected model or reasoning effort. + if (source === "agy-transcript") return metadata; // pi states the provider on the same records that state usage, so the window is asked for // from what this line says rather than remembered across sources let piProvider: string | null = null; diff --git a/server/conversation-monitor.ts b/server/conversation-monitor.ts index 49d8fd777e..91ad186b25 100644 --- a/server/conversation-monitor.ts +++ b/server/conversation-monitor.ts @@ -2,6 +2,7 @@ import { createHash, randomUUID } from "node:crypto"; import { statSync } from "node:fs"; import type { SessionSnapshot } from "../shared/protocol.ts"; import { ConversationNotStarted, ConversationUnavailable, paneTranscript, type NativeTranscript } from "./conversation.ts"; +import { agyDatabasePath } from "./agy.ts"; import { sessionSnapshot } from "./herdr/client.ts"; const STAT_INTERVAL_MS = 250; @@ -81,7 +82,9 @@ export class ConversationMonitor { const wal = database ? fileIdentity(`${transcript.path}-wal`) : ""; const session = transcript.source === "hermes-transcript" ? transcript.sessionId : database ? transcript.session : ""; const cwd = transcript.source === "devin-transcript" ? transcript.cwd : ""; - const observed = `${transcript.source}\0${transcript.path}\0${session}\0${cwd}\0${identity}\0${wal}`; + const catalog = transcript.source === "agy-transcript" ? agyDatabasePath(transcript.path) : null; + const catalogIdentity = catalog === null ? "" : `${fileIdentity(catalog)}:${fileIdentity(`${catalog}-wal`)}`; + const observed = `${transcript.source}\0${transcript.path}\0${session}\0${cwd}\0${identity}\0${wal}\0${catalogIdentity}`; // Resolve once immediately on disappearance, not on every missing-file stat. if (identity === "missing" && observed !== state.observed) state.nextResolve = 0; this.publish(paneId, state, observed); diff --git a/server/conversation.test.ts b/server/conversation.test.ts index 92199f792d..4ff15dbbec 100644 --- a/server/conversation.test.ts +++ b/server/conversation.test.ts @@ -1,17 +1,20 @@ import { afterEach, beforeEach, describe, expect, it, spyOn } from "bun:test"; import { Database } from "bun:sqlite"; -import { appendFileSync, copyFileSync, linkSync, mkdirSync, mkdtempSync, renameSync, rmSync, utimesSync, writeFileSync } from "node:fs"; +import { appendFileSync, copyFileSync, linkSync, mkdirSync, mkdtempSync, renameSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, win32 } from "node:path"; import type { HerdrPane, SessionSnapshot } from "../shared/protocol.ts"; import { MAX_TURNS, parseOmpTranscript } from "./transcript-records.ts"; import { toolVerb } from "../src/lib/toolVerbs.ts"; import type { ConversationTurn } from "../shared/protocol.ts"; +import { parseAgyTranscript } from "./agy.ts"; +import { agyStoreFromLocks } from "./agy-store.ts"; +import * as agyStore from "./agy-store.ts"; import * as herdr from "./herdr/client.ts"; import { forgetHistoryChains } from "./codex.ts"; import { DevinHistoryChanged } from "./devin.ts"; import * as omo from "./omo.ts"; -import { ConversationUnavailable, devinSessionForPane, forgetPaneTranscriptState, forgetTranscriptState, gjcTranscriptPath, HistoryChanged, isDevinProcess, isOmoProcess, ompSessionStore, paneConversation, parseClaudeTranscript, unwrapPastes, transcriptImage, transcriptPage, transcriptToolOutput } from "./conversation.ts"; +import { ConversationUnavailable, devinSessionForPane, forgetPaneTranscriptState, forgetTranscriptState, gjcTranscriptPath, HistoryChanged, isDevinProcess, isOmoProcess, ompSessionStore, paneConversation, paneTranscript, parseClaudeTranscript, unwrapPastes, transcriptImage, transcriptPage, transcriptToolOutput } from "./conversation.ts"; let mockedPanes: HerdrPane[] = []; let mockedProcesses: { argv?: unknown }[] = []; @@ -1467,3 +1470,178 @@ it("expands inherited Codex tool output without exposing output discarded by a b expect(transcriptToolOutput("codex-transcript", leaf, "current", home)).toBe("current output"); } finally { rmSync(home, { recursive: true, force: true }); } }); + +describe("Agy native ledger through shared paging", () => { + let root: string, path: string, db: Database; + const row = (step_index: number, type: string, content = "", extra = {}) => JSON.stringify({ step_index, type, status: "DONE", created_at: "2026-01-01T00:00:00Z", content, ...extra }) + "\n"; + const add = (index: number, type: string, content = "", extra = {}) => { + db.query("INSERT OR REPLACE INTO steps VALUES (?)").run(index); + appendFileSync(path, row(index, type, content, extra)); + }; + const page = () => transcriptPage("agy-transcript", path); + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "herdr-agy-unit-")); + const session = "00000000-0000-4000-8000-000000000001"; + path = join(root, "brain", session, ".system_generated", "logs", "transcript_full.jsonl"); + mkdirSync(join(path, ".."), { recursive: true }); + mkdirSync(join(root, "conversations")); + db = new Database(join(root, "conversations", `${session}.db`)); + db.exec("CREATE TABLE steps(idx INTEGER PRIMARY KEY)"); + writeFileSync(path, ""); + forgetTranscriptState(); + }); + afterEach(() => { db.close(); forgetTranscriptState(); rmSync(root, { recursive: true, force: true }); }); + it("renders native messages, multiple planner calls and independent results without guessed pairing", () => { + add(0, "USER_INPUT", "Keep literal tags\nhidden"); + add(1, "EPHEMERAL_MESSAGE", "hidden context"); + add(2, "PLANNER_RESPONSE", "Checking", { tool_calls: [{ name: "run_command", args: { CommandLine: "echo fiction" } }, { name: "view_file", args: { AbsolutePath: "/fiction.txt" } }] }); + add(3, "GENERIC", "independent output"); + add(4, "CHECKPOINT", "not a result"); + add(5, "USER_INPUT", "next"); + const turns = page().turns; + expect(turns[0]!.parts).toEqual([{ kind: "text", text: "Keep literal tags" }]); + expect(turns[1]!.parts.map((p) => p.kind === "tool" ? [p.name, p.summary, p.output] : p.kind === "text" ? p.text : p.kind)).toEqual([ + "Checking", ["run_command", "echo fiction", ""], ["view_file", "/fiction.txt", ""], ["Tool output", "Step 3", "independent output"], + ]); + expect(turns[2]!.role).toBe("user"); + }); + it("keeps paging and refs stable on append, then rejects them on a DB-only idle rewind", () => { + for (let i = 0; i < 80; i++) { add(i * 2, "USER_INPUT", `prompt ${i}`); add(i * 2 + 1, "GENERIC", "x".repeat(5000)); } + const first = page(); + expect(first.cursor).not.toBeNull(); + const older = transcriptPage("agy-transcript", path, { before: first.cursor! }); + const tool = first.turns.flatMap((t) => t.parts).find((p) => p.kind === "tool")!; + if (tool.kind !== "tool") throw new Error("missing result"); + expect(transcriptToolOutput("agy-transcript", path, tool.output_ref!)).toBe("x".repeat(5000)); + add(160, "USER_INPUT", "append"); + expect(page().history_id).toBe(first.history_id); + expect(transcriptPage("agy-transcript", path, { before: first.cursor! })).toEqual(older); + db.exec("DELETE FROM steps WHERE idx >= 4"); + const rewound = page(); + expect(rewound.turns).toHaveLength(4); + expect(rewound.history_id).not.toBe(first.history_id); + expect(() => transcriptPage("agy-transcript", path, { before: first.cursor! })).toThrow(HistoryChanged); + expect(transcriptToolOutput("agy-transcript", path, tool.output_ref!)).toBeNull(); + forgetTranscriptState(); + expect(page().turns).toEqual(rewound.turns.map((turn) => ({ ...turn, parts: turn.parts.map((part) => part.kind === "tool" ? { ...part, output_ref: expect.any(String) } : part) }))); + }); + it("replays repeated/backward indices and allows database-only rows", () => { + add(0, "USER_INPUT", "one"); add(1, "PLANNER_RESPONSE", "old"); + db.query("INSERT INTO steps VALUES (?)").run(2); + add(3, "USER_INPUT", "two"); add(4, "PLANNER_RESPONSE", "discard"); + const before = page(); + add(3, "USER_INPUT", "replacement"); + expect(page().turns.map((t) => t.parts[0])).toEqual([{ kind: "text", text: "one" }, { kind: "text", text: "old" }, { kind: "text", text: "replacement" }]); + const revision = page().history_id; + expect(revision).not.toBe(before.history_id); + add(4, "PLANNER_RESPONSE", "new"); + expect(page().history_id).toBe(revision); + add(3, "USER_INPUT", "equal repeat"); + expect(page().history_id).not.toBe(revision); + const warm = page().turns; + forgetTranscriptState(); + expect(page().turns).toEqual(warm); + }); + it("ignores an incomplete tail and does not settle a running row", () => { + add(0, "USER_INPUT", "one"); + db.query("INSERT INTO steps VALUES (1)").run(); + const text = row(1, "PLANNER_RESPONSE", "complete"); + appendFileSync(path, text.slice(0, -3)); + expect(page().turns).toHaveLength(1); + appendFileSync(path, text.slice(-3)); + expect(page().turns).toHaveLength(2); + add(2, "PLANNER_RESPONSE", "unfinished", { status: "RUNNING" }); + expect(JSON.stringify(page().turns)).not.toContain("unfinished"); + add(2, "PLANNER_RESPONSE", "finished"); + expect(JSON.stringify(page().turns)).toContain("finished"); + }); + it("admits a ledger tail when the DB catches up without changing older cursors", () => { + add(0, "USER_INPUT", "first"); + const initial = page(); + appendFileSync(path, row(1, "PLANNER_RESPONSE", "lagged")); + expect(page().turns).toHaveLength(1); + db.exec("INSERT INTO steps VALUES (1)"); + const caughtUp = page(); + expect(caughtUp.history_id).toBe(initial.history_id); + expect(caughtUp.turns).toHaveLength(2); + add(2, "PLANNER_RESPONSE", "partial", { status: "RUNNING" }); + const running = page(); + add(2, "PLANNER_RESPONSE", "complete"); + expect(page().history_id).toBe(running.history_id); + expect(transcriptPage("agy-transcript", path, { from: `${initial.history_id}:0` }).turns).toHaveLength(2); + }); + it("never resurrects an observed rewind when replacement DB indices precede their ledger rows", () => { + add(0, "USER_INPUT", "kept"); add(1, "GENERIC", "removed".repeat(1000)); + const before = page(); + db.exec("DELETE FROM steps WHERE idx=1"); + const cut = page(); + db.exec("INSERT INTO steps VALUES (1)"); + expect(page().turns).toEqual(cut.turns); + expect(page().history_id).toBe(cut.history_id); + add(1, "GENERIC", "replacement"); + expect(JSON.stringify(page().turns)).toContain("replacement"); + expect(JSON.stringify(page().turns)).not.toContain("removed"); + const part = before.turns[1]!.parts[0]!; + if (part.kind !== "tool") throw new Error("missing tool"); + expect(transcriptToolOutput("agy-transcript", path, part.output_ref!)).toBeNull(); + }); + it("does not hide later turns behind an unfinished or unknown-status historical row", () => { + add(0, "USER_INPUT", "first"); + add(1, "GENERIC", "unfinished", { status: "UNKNOWN" }); + expect(page().turns).toHaveLength(1); + add(2, "USER_INPUT", "later"); add(3, "PLANNER_RESPONSE", "answer"); + expect(page().turns.map(t => t.parts[0])).toEqual([ + { kind: "text", text: "first" }, { kind: "text", text: "later" }, { kind: "text", text: "answer" }, + ]); + }); + it("uses the same assistant boundaries in full and paged replay around a skipped user row", () => { + add(0, "USER_INPUT", "first"); add(1, "PLANNER_RESPONSE", "before"); + add(2, "USER_INPUT", "hidden", { status: "UNKNOWN" }); + add(3, "PLANNER_RESPONSE", "after"); + const full = parseAgyTranscript(row(0, "USER_INPUT", "first") + row(1, "PLANNER_RESPONSE", "before") + + row(2, "USER_INPUT", "hidden", { status: "UNKNOWN" }) + row(3, "PLANNER_RESPONSE", "after")); + expect(page().turns).toEqual(full); + expect(full).toHaveLength(2); + }); + it("continues a bounded cold scan on the next poll", () => { + add(0, "USER_INPUT", "start"); + for (let i=1; i<=34; i++) add(i, "EPHEMERAL_MESSAGE", "x".repeat(1024 * 1024)); + add(35, "USER_INPUT", "end"); + expect(page).toThrow(ConversationUnavailable); + expect(page().turns.at(-1)!.parts[0]).toEqual({ kind: "text", text: "end" }); + }); + it("preserves standalone literal user wrapper text", () => { + add(0, "USER_INPUT", "literal"); + expect(page().turns[0]!.parts[0]).toEqual({ kind: "text", text: "literal" }); + }); + it.skipIf(process.platform === "win32")("canonicalizes held locks and never chooses among sessions or unheld files", () => { + const presence = join(root, "presence"); mkdirSync(presence); + const one = join(presence, "00000000-0000-4000-8000-000000000001.lock"); + const two = join(presence, "00000000-0000-4000-8000-000000000002.lock"); + writeFileSync(one, ""); writeFileSync(two, ""); + expect(agyStoreFromLocks([])).toBeNull(); + expect(agyStoreFromLocks([one, two])).toBeNull(); + expect(agyStoreFromLocks([one])?.path).toBe(path); + const alias = join(root, "alias"); symlinkSync(presence, alias); + // Alias a parent of presence, preserving the native held-path shape. + const link = join(root, "linked-root"); symlinkSync(root, link); + expect(agyStoreFromLocks([join(link, "presence", one.split("/").at(-1)!)])?.path).toBe(path); + }); + it("distinguishes unreadable database from verified empty history", () => { + add(0, "USER_INPUT", "one"); + page(); + db.exec("DROP TABLE steps"); + expect(page).toThrow(ConversationUnavailable); + }); + it("falls back when the transcript path cannot be inspected", async () => { + const loop = join(root, "loop"); symlinkSync(loop, loop); + const pane = { pane_id: "agy-pane", cwd: root, agent: "agy" } as HerdrPane; + const store = spyOn(agyStore, "resolveAgyStore").mockResolvedValue({ root, session: "test", path: loop, pid: 2, started: "test" }); + const rpc = spyOn(herdr, "herdrRpc").mockResolvedValue({ process_info: { foreground_processes: [] } } as never); + try { + await expect(paneTranscript(pane.pane_id, undefined, { panes: [pane] } as SessionSnapshot)).rejects.toMatchObject({ + name: "ConversationUnavailable", message: "transcript_missing", + }); + } finally { store.mockRestore(); rpc.mockRestore(); } + }); +}); diff --git a/server/conversation.ts b/server/conversation.ts index 9dee6ef4c1..f6048aacf2 100644 --- a/server/conversation.ts +++ b/server/conversation.ts @@ -52,6 +52,8 @@ import { forgetAllPiIndexes, forgetPiIndex, piAbandonedTurns, piBranchSegments } import { defaultDevinDbPath, DevinHistoryUnavailable, devinConversation, forgetDevinState } from "./devin.ts"; import { forgetGrokState, grokProjection, grokToolOutput, parseGrokTranscript, GrokUnavailable } from "./grok.ts"; import { resolveGrokStore } from "./grok-store.ts"; +import { AgyUnavailable, agyCatalog, agyToolOutput, forgetAgyState, parseAgyTranscript, withAgyHistory, type AgyProjection } from "./agy.ts"; +import { resolveAgyStore, sameAgyStore, type AgyStore } from "./agy-store.ts"; import { trimOutput } from "./tool-output.ts"; import { parseConversationMetadata } from "./conversation-metadata.ts"; import { forgetHermesTranscriptState, hermesConversationPage, hermesToolOutput, hermesTranscriptForPane } from "./hermes.ts"; @@ -460,7 +462,7 @@ export class HistoryChanged extends Error { /** What paneConversation resolved: which store the turns came from, and where they start. */ export type RecognizedConversation = { - source: "grok-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript"; + source: "grok-transcript" | "agy-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript"; turns: ConversationTurn[]; metadata: ConversationMetadata; /** the first turn's position, for the page before it; null at the conversation's beginning */ @@ -506,6 +508,7 @@ interface TranscriptStream { length: number; floor: number; promptStarts?: number[]; + fd?: number; } // In-place rewrites keep the inode. Change the cursor generation when observed, @@ -521,7 +524,7 @@ function transcriptGeneration(path: string, stat: { dev: number; ino: number; si return generation ? `-${generation}` : ""; } -function transcriptStream(source: StreamSource, path: string, stat: { dev: number; ino: number; size: number; mtimeMs: number; ctimeMs: number }, codexHome: string): TranscriptStream { +function transcriptStream(source: StreamSource, path: string, stat: { dev: number; ino: number; size: number; mtimeMs: number; ctimeMs: number }, codexHome: string, agy?: AgyProjection, fd?: number): TranscriptStream { // (a chain whose parent was archived since comes back shorter: codexHistorySegments; // a pi branch skips the side paths a /tree left behind: piBranchSegments) const grok = source === "grok-transcript" ? grokProjection(path, stat) : null; @@ -530,6 +533,7 @@ function transcriptStream(source: StreamSource, path: string, stat: { dev: numbe const segments: { path: string; start: number; end: number }[] = source === "codex-transcript" ? codexHistorySegments(path, codexHome).map((segment) => ({ path: segment.path, start: 0, end: segment.end })) : grok !== null ? grok.segments.map((segment) => ({ path, ...segment })) + : agy ? agy.segments.map((segment) => ({ path, ...segment })) : branch !== null ? branch.map((segment) => ({ path, start: segment.start, end: segment.end })) : [{ path, start: 0, end: stat.size }]; const files: TranscriptStream["files"] = []; @@ -554,6 +558,8 @@ function transcriptStream(source: StreamSource, path: string, stat: { dev: numbe let chain = ""; if (grok !== null) { chain = `-grok-${createHash("sha256").update(path).digest("base64url").slice(0, 10)}-${grok.revision}`; + } else if (agy) { + chain = `-agy-${agy.revision}`; } else if (branch !== null) { const layout = files.map((file, at) => (at === files.length - 1 ? `${file.offset}` : `${file.offset}\0${file.length}`)).join("\n"); chain = `-${createHash("sha256").update(layout).digest("base64url").slice(0, 10)}`; @@ -564,7 +570,7 @@ function transcriptStream(source: StreamSource, path: string, stat: { dev: numbe }); chain = earlier.length === 0 ? "" : `-${createHash("sha256").update(earlier.join("\n")).digest("base64url").slice(0, 10)}`; } - return { id: `${stat.dev.toString(36)}-${stat.ino.toString(36)}${chain}${transcriptGeneration(path, stat)}`, files, length: stream, floor: 0, promptStarts: grok?.promptStarts }; + return { id: `${stat.dev.toString(36)}-${stat.ino.toString(36)}${chain}${transcriptGeneration(path, stat)}`, files, length: stream, floor: 0, promptStarts: grok?.promptStarts ?? agy?.promptStarts, fd }; } function readStream(stream: TranscriptStream, from: number, to: number): Buffer { @@ -573,12 +579,12 @@ function readStream(stream: TranscriptStream, from: number, to: number): Buffer const low = Math.max(from, file.start); const high = Math.min(to, file.start + file.length); if (low >= high) continue; - const fd = openSync(file.path, "r"); + const fd = stream.fd ?? openSync(file.path, "r"); try { const buffer = Buffer.alloc(high - low); chunks.push(buffer.subarray(0, readSync(fd, buffer, 0, buffer.length, file.offset + (low - file.start)))); } finally { - closeSync(fd); + if (stream.fd === undefined) closeSync(fd); } } return Buffer.concat(chunks); @@ -592,7 +598,7 @@ function readStream(stream: TranscriptStream, from: number, to: number): Buffer const SETTING_TYPES = ['"model_change"', '"thinking_level_change"']; const clearScans = new Map> }>(); function applyHistoryBoundary(path: string, stream: TranscriptStream, source: RecognizedConversation["source"]): void { - if (source === "codex-transcript" || source === "grok-transcript") return; + if (source === "codex-transcript" || source === "grok-transcript" || source === "agy-transcript") return; let scan = clearScans.get(path); if (!scan || scan.id !== stream.id || scan.scanned > stream.length || bytesBefore(stream, scan.scanned) !== scan.tail) { scan = { id: stream.id, scanned: 0, floor: 0, tail: "", settings: {} }; @@ -637,6 +643,7 @@ function applyHistoryBoundary(path: string, stream: TranscriptStream, source: Re /** Bytes that every line opening a turn contains: a cheap filter before JSON.parse. */ const TURN_MARK: Record = { "grok-transcript": Buffer.from('"user_message_chunk"'), + "agy-transcript": Buffer.from('"USER_INPUT"'), "codex-transcript": Buffer.from('"task_started"'), "claude-transcript": Buffer.from('"user"'), "omp-transcript": Buffer.from('"user"'), @@ -866,6 +873,7 @@ function parseTurns(source: RecognizedConversation["source"], path: string, text throw error; } } + if (source === "agy-transcript") return parseAgyTranscript(text, path); return source === "codex-transcript" ? parseCodexTranscript(text, Infinity) // only pi keeps a tool's images in the entry as base64; omp, omo and gjc are read the same // way but would carry image refs nothing can answer, so the option stays with pi alone @@ -985,6 +993,7 @@ export function forgetTranscriptState(): void { forgetAllCodexState(); forgetAllPiIndexes(); forgetGrokState(); + forgetAgyState(); forgetDevinState(); forgetHermesTranscriptState(); } @@ -1019,6 +1028,7 @@ export function forgetPaneTranscriptState(paneId: string): void { forgetCodexStateFor(path); forgetPiIndex(path); forgetGrokState(path); + forgetAgyState(path); forgetClaudeSessionFile(path); forgetOpencodeRead(path); forgetHermesTranscriptState(path); @@ -1178,7 +1188,7 @@ async function ompTranscriptPath(paneId: string, cwd: string): Promise { /** Where a pane's conversation is: a transcript file, or a session in OpenCode's database. */ type ResolvedTranscript = - | { source: StreamSource; path: string; codexHome?: string } + | { source: StreamSource; path: string; codexHome?: string; agy?: AgyStore } | { source: "opencode-transcript"; path: string; session: string } | { source: "hermes-transcript"; path: string; sessionId: string }; @@ -1223,6 +1233,18 @@ async function resolveTranscript(pane: HerdrPane, cwd: string, codexHome?: strin if (!found) throw new ConversationUnavailable("no_session_path"); return { source: "grok-transcript", path: found }; } + if (agent === "agy") { + const found = await resolveAgyStore(pane); + if (!found) throw new ConversationUnavailable("no_session_path"); + let transcriptExists: boolean; + try { transcriptExists = !!statSync(found.path, { throwIfNoEntry: false }); } + catch { throw new ConversationUnavailable("transcript_missing"); } + if (!transcriptExists) { + if (agyCatalog(found.path).size !== 0 || !sameAgyStore(found, await resolveAgyStore(pane))) throw new ConversationUnavailable("transcript_missing"); + throw new ConversationNotStarted(found.session, "agy-transcript", found.path); + } + return { source: "agy-transcript", path: found.path, agy: found }; + } if (agent === "codex") { // the pane's own store: the rollout, its history and its images are all read from there const home = await paneCodexHome(paneId, codexHome); @@ -1255,6 +1277,7 @@ async function resolveTranscript(pane: HerdrPane, cwd: string, codexHome?: strin } throw new ConversationUnavailable("no_recognized_transcript"); } catch (error) { + if (error instanceof AgyUnavailable) throw new ConversationUnavailable("transcript_missing"); if (!(error instanceof ConversationUnavailable) || !(await paneRunsOmo(paneId))) throw error; return { source: "omo-transcript", path: await omoTranscriptPath(paneId, cwd, panes) }; } @@ -1406,7 +1429,8 @@ export async function paneConversation(paneId: string, codexHome?: string, page: throw error; } } - const identity = resolved.source === "claude-transcript" ? nodePath.basename(resolved.path, ".jsonl") + const identity = resolved.source === "agy-transcript" ? resolved.path + : resolved.source === "claude-transcript" ? nodePath.basename(resolved.path, ".jsonl") : resolved.source === "pi-transcript" || resolved.source === "omp-transcript" ? realpathSync(resolved.path) : null; rememberPaneRead(paneId, resolved.path); const answer = transcriptPage(resolved.source, resolved.path, page, resolved.codexHome ?? codexHome); @@ -1418,12 +1442,24 @@ export async function paneConversation(paneId: string, codexHome?: string, page: try { revision = grokProjection(resolved.path, statSync(resolved.path)).revision; } catch { /* unreadable now */ } if (revision === null || revision === "0") throw new ConversationUnavailable("transcript_missing"); } + if (resolved.agy && !sameAgyStore(resolved.agy, await resolveAgyStore(snapshot.panes.find((pane) => pane.pane_id === paneId)!))) throw new ConversationUnavailable("no_session_path"); if (identity !== null) { writtenSessions.add(`${resolved.source}\0${identity}`); rememberPaneSession(paneId, `${resolved.source}\0${identity}`); } return answer; } /** One page of a resolved transcript (paneConversation's `page`). */ export function transcriptPage(source: StreamSource, path: string, page: ConversationPage = {}, codexHome?: string): RecognizedConversation { + if (source === "agy-transcript") { + try { return withAgyHistory(path, (projection, fd) => transcriptPageRead(source, path, page, codexHome, projection, fd)); } + catch (error) { + if (error instanceof HistoryChanged || error instanceof ConversationUnavailable) throw error; + throw new ConversationUnavailable("transcript_missing"); + } + } + return transcriptPageRead(source, path, page, codexHome); +} + +function transcriptPageRead(source: StreamSource, path: string, page: ConversationPage, codexHome?: string, agy?: AgyProjection, fd?: number): RecognizedConversation { let stat: { dev: number; ino: number; size: number; mtimeMs: number; ctimeMs: number }; try { stat = statSync(path); @@ -1432,7 +1468,7 @@ export function transcriptPage(source: StreamSource, path: string, page: Convers } let stream: TranscriptStream; try { - stream = transcriptStream(source, path, stat, codexHome ?? defaultCodexHome()); + stream = transcriptStream(source, path, stat, codexHome ?? defaultCodexHome(), agy, fd); applyHistoryBoundary(path, stream, source); } catch (error) { // a branch that cannot be walked says so; a file that went unreadable mid-read has @@ -1443,7 +1479,7 @@ export function transcriptPage(source: StreamSource, path: string, page: Convers // an older page never changes while its file and the rollouts before it stay the same // (the stream's id names both); the newest one changes with every append const key = page.before !== undefined ? `${path}\0before:${page.before}:${page.since ?? ""}` : `${path}\0from:${page.from ?? ""}`; - const signature = page.before !== undefined ? stream.id : `${stream.id}:${stat.size}:${stat.mtimeMs}:${stat.ctimeMs}`; + const signature = page.before !== undefined ? stream.id : `${stream.id}:${stream.length}:${stat.size}:${stat.mtimeMs}:${stat.ctimeMs}`; const cached = cache.get(key); // the answer is a function of the page asked for and the file's state, so they name it const version = answerVersion(key, signature); @@ -1571,7 +1607,9 @@ export async function toolOutput(paneId: string, ref: string, codexHome?: string } rememberPaneRead(paneId, resolved.path); if (resolved.source === "hermes-transcript") return hermesToolOutput(resolved.sessionId, resolved.path, ref, TOOL_OUTPUT_MAX); - return transcriptToolOutput(resolved.source, resolved.path, ref, resolved.codexHome ?? codexHome); + const output = transcriptToolOutput(resolved.source, resolved.path, ref, resolved.codexHome ?? codexHome); + if (resolved.agy && !sameAgyStore(resolved.agy, await resolveAgyStore(pane))) return null; + return output; } /** The output a transcript file holds for one tool call id, whole (up to TOOL_OUTPUT_MAX). */ @@ -1590,6 +1628,7 @@ export function transcriptToolOutput(source: StreamSource, path: string, ref: st return null; } if (source === "grok-transcript") return grokToolOutput(path, ref); + if (source === "agy-transcript") return agyToolOutput(path, ref); if (source === "pi-transcript") return piToolOutput(path, ref); let text: string; try { text = readFileSync(path, "utf8"); } catch { return null; } diff --git a/shared/protocol.ts b/shared/protocol.ts index bc3ecfd62e..a9ba2a539a 100644 --- a/shared/protocol.ts +++ b/shared/protocol.ts @@ -107,9 +107,10 @@ export type { Machine, MachineEvent, PaneTarget, SetupJob, SetupRequest, SetupAc * history, soft-wrapped lines joined; a terminal selection that outlives one screen) * POST /api/pane/input { pane_id, text } -> { ok: true } * GET /api/pane/conversation?pane_id= -> ConversationResponse (structured agent - * transcript turns - claude, codex, omp, omo, gjc, pi, devin, OpenCode, hermes or grok; + * transcript turns - claude, codex, omp, omo, gjc, pi, devin, OpenCode, hermes, grok or agy; * source:"scrollback" when the pane has no recognized, readable store. - * Grok requires exact live process/session evidence; rewinds invalidate cursors.) + * Grok requires exact live process/session evidence, Agy live process/session evidence; + * rewinds invalidate cursors.) * POST /api/pane/close { pane_id } -> { ok: true } (pane.close RPC; the collector's * session-changed broadcast removes it from every client's sidebar) * POST /api/pane/rename { pane_id, label } -> { ok: true } (pane.rename; empty label clears it) @@ -507,7 +508,7 @@ export interface PlanSummary { export interface ConversationResponse { /** Stable across appends; changes on transcript replacement or native context clear. */ history_id?: string; - source: "grok-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript" | "scrollback"; + source: "grok-transcript" | "agy-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript" | "scrollback"; turns: ConversationTurn[]; metadata?: ConversationMetadata; /** diff --git a/src/components/ChatView.tsx b/src/components/ChatView.tsx index 70a4555696..2d9f2bc0e6 100644 --- a/src/components/ChatView.tsx +++ b/src/components/ChatView.tsx @@ -43,7 +43,7 @@ import { sendUserFileCall } from "../lib/sendUserFile.ts"; const ChatPaneContext = createContext(null); const ChatHistoryContext = createContext(""); import type { TypedAnswer } from "../lib/promptAnswer.ts"; -import type { AgentStatus, ConversationMetadata, ConversationPart, ConversationTurn, InteractivePrompt, OmoTaskResult } from "../../shared/protocol.ts"; +import type { AgentStatus, ConversationMetadata, ConversationPart, ConversationResponse, ConversationTurn, InteractivePrompt, OmoTaskResult } from "../../shared/protocol.ts"; import { chatIsBlank, type ChatRead } from "../lib/greeting.ts"; import { currentLocale, useT } from "../lib/i18n.ts"; @@ -589,7 +589,7 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, invalidatio // polls pause while the page is hidden and pick up at once when it is back const visible = usePageVisible(); /** the answer last laid out: an unchanged poll (a 304) hands back this very object */ - const lastAnswer = useRef(null); + const lastAnswer = useRef(null); const [conversationRefresh] = useState(() => new ConversationRefresh()); const onMetadataRef = useRef(onMetadata); onMetadataRef.current = onMetadata; @@ -748,6 +748,13 @@ export const ChatView = memo(function ChatView({ paneId, refreshKey, invalidatio if (conversation.source !== "scrollback") next = { source: "conversation", conversationSource: conversation.source, turns: conversation.turns, messages: [], truncated: false }; else if (agent !== null) next = { source: "scrollback", turns: [], messages: [], truncated: false }; else { + // The reader has rejected this history. Hide it before the separate terminal + // request, which may wait or fail; a transport error must not revive those turns. + if (lastAnswer.current?.source !== "scrollback") { + setState((previous) => previous.source === "conversation" + ? { source: "scrollback", turns: [], messages: [], truncated: false } : previous); + signature.current = ""; lastAnswer.current = null; + } const result = await fetchPaneTranscript(paneId, TRANSCRIPT_LINES); if (cancelled || generation !== olderGeneration.current) return; next = { source: "scrollback", turns: [], messages: toTranscriptMessages(result.text).filter((message) => message.role !== "status"), truncated: result.truncated === true };