diff --git a/changelog.d/letta-native-chat.md b/changelog.d/letta-native-chat.md new file mode 100644 index 0000000000..5627df7886 --- /dev/null +++ b/changelog.d/letta-native-chat.md @@ -0,0 +1,4 @@ +### Added +- Chat shows a Letta Code pane's conversation, read from Letta's own transcript, when herdr's Letta + integration reports it and Letta's own record names the same conversation; otherwise the + terminal's text stays. diff --git a/docs/guide.md b/docs/guide.md index 8c9b76c10f..55fdc82d2e 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -169,7 +169,7 @@ Every agent herdr runs shows up with its live status, terminal and alerts. The c | **Hermes** | Native SQLite session store | — use Terminal | | **Muse Code** | **Linux untested with real Muse.** Native session log, bound to the pane by the session lock Muse writes for its process; panes in one folder each show their own session. macOS and Linux; Windows uses Terminal. See [Muse chat](muse-chat.md). | — 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 | -| **Letta Code** | The terminal's text. Choose **Letta Code** in **New workspace**, **New tab with agent…** or **New worktree** where `letta` is installed, or run `letta` in a herdr pane's **Terminal**. | — use Terminal | +| **Letta Code** | **Experimental; Linux and macOS.** Read-only native transcript from where the pane's own `letta` process keeps it (`~/.letta/transcripts`, or its `LETTA_CONFIG_DIR` / `LETTA_TRANSCRIPT_ROOT`), when herdr's Letta integration is installed (`herdr integration install letta`, experimental), it names a conversation for the pane, and Letta's own record in the project folder (`.letta/settings.local.json`) names the same one, is older than that `letta` process and is writable by it, and `letta` is still in the folder it started in (for a local-backend conversation, Letta's local store must also name that agent for it). That holds when `letta` resumes the conversation it last had in that folder (`letta`, or `letta --conversation ` on that one). After `/new` or `/resume`, a start on another conversation (`--new` included), once `letta` has moved to another folder (its SetWorkingDirectory or EnterWorktree tools), another `letta` writing that record, while another pane runs `letta` in the same folder, with two panes on one conversation, in the home folder, or before the conversation's first turn on this PC ends, Chat uses the terminal's text until `letta` starts again on that conversation. Letta writes a turn when it ends: a turn appears once it is done, and a cancelled one never does. Only turns Letta Code wrote on this PC show. Choose **Letta Code** in **New workspace**, **New tab with agent…** or **New worktree** where `letta` is installed, or run `letta` in a herdr pane's **Terminal**. Windows uses Terminal. | — use Terminal | | **GitHub Copilot CLI** | Linux native transcript when the foreground session is verified. See [Copilot setup and limits](copilot-chat.md). | — recorded choices are read-only; answer questions and approvals in Terminal | | **Anything else** | The terminal's text | — use Terminal | diff --git a/server/conversation-metadata.ts b/server/conversation-metadata.ts index 1498237d38..4ec78d5efe 100644 --- a/server/conversation-metadata.ts +++ b/server/conversation-metadata.ts @@ -84,6 +84,8 @@ export function parseConversationMetadata(text: string, source: ConversationResp 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; + // Letta's transcript names no model or effort + if (source === "letta-transcript") return metadata; if (source === "copilot-transcript") { for (const { type, data } of copilotRecords(text)) { const model = type === "session.start" ? label(data.selectedModel) diff --git a/server/conversation-shared.ts b/server/conversation-shared.ts index 56fa946c52..728d063aba 100644 --- a/server/conversation-shared.ts +++ b/server/conversation-shared.ts @@ -53,7 +53,7 @@ export class HistoryChanged extends Error { /** What paneConversation resolved: which store the turns came from, and where they start. */ export type RecognizedConversation = { activity?: "thinking" | "working" | "idle"; - source: "copilot-transcript" | "grok-transcript" | "agy-transcript" | "muse-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "cmd-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript"; + source: "copilot-transcript" | "grok-transcript" | "agy-transcript" | "muse-transcript" | "letta-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "cmd-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript"; turns: ConversationTurn[]; metadata: ConversationMetadata; /** `metadata` was read off the agent's own screen (ConversationResponse.screen) */ diff --git a/server/conversation.ts b/server/conversation.ts index ebb1d8c849..18be2507c5 100644 --- a/server/conversation.ts +++ b/server/conversation.ts @@ -60,6 +60,8 @@ import { AgyUnavailable, agyCatalog, agyToolOutput, forgetAgyState, parseAgyTran import { resolveAgyStore, sameAgyStore, type AgyStore } from "./agy-store.ts"; import { museSession, museToolOutput, MuseUnavailable, parseMuseTranscript } from "./muse.ts"; import { resolveMuseSession } from "./muse-store.ts"; +import { LettaUnavailable, lettaToolOutput, parseLettaTranscript } from "./letta.ts"; +import { resolveLettaTranscript } from "./letta-store.ts"; import { withinRetentionBudget } from "./retention-budget.ts"; import { forgetAllPiIndexes, forgetPiIndex, piAbandonedTurns, piBranchRecords, piBranchSegments } from "./pi-tree.ts"; import { copilotOffsetOutput, copilotOutput, copilotRootRecord, parseCopilotTranscript } from "./copilot.ts"; @@ -302,7 +304,7 @@ function applyHistoryBoundary(path: string, stream: TranscriptStream, source: Re * what is kept and puts it back when it ends, so a page read between its steps never sees half of it. */ function* applyHistoryBoundarySteps(path: string, stream: TranscriptStream, source: RecognizedConversation["source"], apart: boolean): Steps { - if (source === "codex-transcript" || source === "grok-transcript" || source === "agy-transcript" || source === "muse-transcript") return; + if (source === "codex-transcript" || source === "grok-transcript" || source === "agy-transcript" || source === "muse-transcript" || source === "letta-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: {}, ...(source === "claude-transcript" ? { branch: { last: null, ends: new Map(), answers: [], forks: [] } } : {}) }; @@ -555,6 +557,12 @@ function parseTurns(source: RecognizedConversation["source"], path: string, text throw error; } } + if (source === "letta-transcript") { + try { return parseLettaTranscript(text); } catch (error) { + if (error instanceof LettaUnavailable) throw new ConversationUnavailable("transcript_missing"); + throw error; + } + } return source === "codex-transcript" ? parseCodexTranscript(text, Infinity) : source === "claude-transcript" ? parseClaudeTranscript(text, Infinity, { subagents: subagentsOnPage(path, text), noticed }) : parseOmpTranscript(text, Infinity, ompReading(source, path, taskTitles)); @@ -1274,8 +1282,8 @@ async function resolveTranscript(pane: HerdrPane, cwd: string, codexHome?: strin ); if (running || await gjcPidUnderShell(info?.process_info?.shell_pid)) agent = "gjc"; } - // a Muse pane's processes are asked for once, for omo and for its session lock - if (agent === "muse" && known === undefined) known = (await paneProcessInfo(paneId).catch(() => undefined)) ?? null; + // a Muse or Letta pane's processes are asked for once, for omo and for its own binding + if ((agent === "muse" || agent === "letta") && known === undefined) known = (await paneProcessInfo(paneId).catch(() => undefined)) ?? null; if (await paneRunsOmo(paneId, known)) { return { source: "omo-transcript", path: await omoTranscriptPath(paneId, cwd, panes) }; } @@ -1294,6 +1302,13 @@ async function resolveTranscript(pane: HerdrPane, cwd: string, codexHome?: strin if (path === null) throw new ConversationUnavailable("no_session_path"); return { source: "muse-transcript", path }; } + // herdr's label, not a report an earlier Letta left behind: the binding needs a running one + if (agent === "letta" && pane.agent === "letta") { + // no file yet is no proof the conversation has no turns (one begun elsewhere): Terminal stands in + const found = await resolveLettaTranscript(pane, panes ?? [pane], known); + if (found === null) throw new ConversationUnavailable("no_session_path"); + return { source: "letta-transcript", path: found.path }; + } if (agent === "agy") { const found = await resolveAgyStore(pane); if (!found) throw new ConversationUnavailable("no_session_path"); @@ -1523,7 +1538,7 @@ export async function paneConversation(paneId: string, codexHome?: string, page: throw error; } } - const identity = resolved.source === "agy-transcript" || resolved.source === "copilot-transcript" ? resolved.path + const identity = resolved.source === "agy-transcript" || resolved.source === "copilot-transcript" || resolved.source === "letta-transcript" ? resolved.path : resolved.source === "claude-transcript" ? nodePath.basename(resolved.path, ".jsonl") : resolved.source === "pi-transcript" || resolved.source === "cmd-transcript" || resolved.source === "omp-transcript" ? realpathSync(resolved.path) : null; rememberPaneRead(paneId, resolved.path); @@ -2272,6 +2287,10 @@ export function transcriptToolOutput(source: StreamSource, path: string, ref: st const output = museToolOutput(path, ref); return output !== null && output.length > TOOL_OUTPUT_MAX ? `${output.slice(0, TOOL_OUTPUT_MAX)}\n… trimmed` : output; } + if (source === "letta-transcript") { + const output = lettaToolOutput(path, ref); + return output !== null && output.length > TOOL_OUTPUT_MAX ? `${output.slice(0, TOOL_OUTPUT_MAX)}\n… trimmed` : output; + } if (source === "pi-transcript" || source === "cmd-transcript") return piToolOutput(path, ref); if (source === "copilot-transcript") { let fd: number; diff --git a/server/fixtures/letta/local-conversation-0.34.27.json b/server/fixtures/letta/local-conversation-0.34.27.json new file mode 100644 index 0000000000..00060135eb --- /dev/null +++ b/server/fixtures/letta/local-conversation-0.34.27.json @@ -0,0 +1,18 @@ +{ + "id": "local-conv-1", + "agent_id": "agent-local-97f4bceb-2f0d-4a16-8b9f-abb2f3d4a72c", + "archived": false, + "archived_at": null, + "created_at": "2026-10-29T01:55:23.137Z", + "updated_at": "2026-10-29T01:59:01.109Z", + "last_message_at": "2026-10-29T01:59:01.109Z", + "summary": "Reply with exactly: second conversation ok", + "in_context_message_ids": [ + "ui-msg-10", + "ui-msg-11", + "ui-msg-12", + "ui-msg-13", + "ui-msg-14", + "ui-msg-15" + ] +} diff --git a/server/fixtures/letta/settings.local-0.34.27.json b/server/fixtures/letta/settings.local-0.34.27.json new file mode 100644 index 0000000000..b8a0ad1efd --- /dev/null +++ b/server/fixtures/letta/settings.local-0.34.27.json @@ -0,0 +1,13 @@ +{ + "lastAgent": "agent-local-97f4bceb-2f0d-4a16-8b9f-abb2f3d4a72c", + "sessionsByServer": { + "local:/home/user/.letta/lc-local-backend": { + "agentId": "agent-local-97f4bceb-2f0d-4a16-8b9f-abb2f3d4a72c", + "conversationId": "local-conv-1" + } + }, + "lastSession": { + "agentId": "agent-local-97f4bceb-2f0d-4a16-8b9f-abb2f3d4a72c", + "conversationId": "local-conv-1" + } +} \ No newline at end of file diff --git a/server/fixtures/letta/transcript-conversation-0.34.27.jsonl b/server/fixtures/letta/transcript-conversation-0.34.27.jsonl new file mode 100644 index 0000000000..38cdc1162d --- /dev/null +++ b/server/fixtures/letta/transcript-conversation-0.34.27.jsonl @@ -0,0 +1,5 @@ +{"kind":"user","text":"Reply with exactly: second conversation ok","captured_at":"2026-10-29T01:55:46.561Z","source_line_id":"user-mvsw60vy-gtmdpv"} +{"kind":"assistant","text":"second conversation ok","captured_at":"2026-10-29T01:55:46.561Z","source_line_id":"ui-msg-11","source_message_id":"ui-msg-11"} +{"kind":"user","text":"Use the shell tool to run: cat missing-file.txt. Then tell me in one sentence what happened.","captured_at":"2026-10-29T01:59:01.253Z","source_line_id":"user-mvswa5bi-bfsv6p"} +{"kind":"tool_call","name":"Bash","argsText":"{\"command\":\"cat missing-file.txt\",\"description\":\"Read missing file\"}","resultText":"Exit code: 1\ncat: missing-file.txt: No such file or directory\n","resultOk":false,"captured_at":"2026-10-29T01:59:01.253Z","source_line_id":"Bash-1793239139187262013-3"} +{"kind":"assistant","text":"When I ran the command to read the missing file, it failed because the file does not exist.","captured_at":"2026-10-29T01:59:01.253Z","source_line_id":"ui-msg-15","source_message_id":"ui-msg-15"} diff --git a/server/fixtures/letta/transcript-default-0.34.27.jsonl b/server/fixtures/letta/transcript-default-0.34.27.jsonl new file mode 100644 index 0000000000..8b0acebb5e --- /dev/null +++ b/server/fixtures/letta/transcript-default-0.34.27.jsonl @@ -0,0 +1,6 @@ +{"kind":"user","text":"Use your shell tool to run: echo fixture-ok && ls. Then reply in one short sentence.","captured_at":"2026-10-29T01:53:05.293Z","source_line_id":"user-mvsw2ivj-7t557v"} +{"kind":"tool_call","name":"Bash","argsText":"{\"command\":\"echo fixture-ok && ls\",\"description\":\"Run test command to check fixture and list files\"}","resultText":"fixture-ok\n","resultOk":true,"captured_at":"2026-10-29T01:53:05.293Z","source_line_id":"Bash-1793238783145231402-1"} +{"kind":"assistant","text":"I have successfully run the command and verified the environment.","captured_at":"2026-10-29T01:53:05.293Z","source_line_id":"ui-msg-5","source_message_id":"ui-msg-5"} +{"kind":"user","text":"Read notes.txt and tell me its second line.","captured_at":"2026-10-29T01:55:05.021Z","source_line_id":"user-mvsw53jx-6tdo87"} +{"kind":"tool_call","name":"Read","argsText":"{\"file_path\":\"/home/user/proj/notes.txt\"}","resultText":"1\talpha\n2\tbeta\n3\t","resultOk":true,"captured_at":"2026-10-29T01:55:05.021Z","source_line_id":"Read-1793238903065214259-2"} +{"kind":"assistant","text":"The second line of notes.txt is \"beta\".","captured_at":"2026-10-29T01:55:05.021Z","source_line_id":"ui-msg-9","source_message_id":"ui-msg-9"} diff --git a/server/letta-store.ts b/server/letta-store.ts new file mode 100644 index 0000000000..e34382bc33 --- /dev/null +++ b/server/letta-store.ts @@ -0,0 +1,294 @@ +/** + * Which Letta Code conversation a pane shows. Three facts must agree, or Chat uses Terminal: + * - herdr's report from its Letta integration (`herdr integration install letta`): `herdr:letta`, + * the conversation id, or `default:` for an agent's default conversation. herdr 0.9.5 + * keeps the first report of a letta process, so after `/new` or `/resume` it still names the + * conversation the pane left. + * - Letta's own record of the conversation it opened, which every letta process writes into its + * project's `.letta/settings.local.json` (`sessionsByServer`) on start and on every switch, and + * reads back to resume. It is shared by every letta in that folder, and nothing in it says which + * process wrote it, so it speaks for the pane's process only when it is older than that process: + * a letta that opens the conversation the record names writes nothing (`setLocalLastSession` + * skips an equal session), and any later write (a `/new` or `/resume`, its own start on another + * conversation, or another letta, even one gone since) leaves Chat on Terminal until letta starts again. + * - The pane's one letta process: its own environment for where Letta keeps its files + * (`LETTA_CONFIG_DIR`, `LETTA_TRANSCRIPT_ROOT`, `HOME`), read from the process on Linux and macOS + * (Windows has no such read, and uses Terminal), and the folder it started in (lettaInStartFolder), + * whose record is the only one read, while the process is still there. + * The transcript must then be a regular file at exactly the path Letta derives from the two ids. + */ +import { accessSync, closeSync, constants, fstatSync, lstatSync, openSync, readFileSync, readSync, realpathSync } from "node:fs"; +import { basename, isAbsolute, join, normalize } from "node:path"; +import type { HerdrPane } from "../shared/protocol.ts"; +import { darwinProcess } from "./claude-store.ts"; +import { paneProcessInfo, type ProcessInfo } from "./herdr/client.ts"; +import { processStartLowerBound } from "./process-start.ts"; + +/** Letta's ids: `conv-`, `local-conv-`, `agent-`, `agent-local-`, and `default`. */ +const ID = /^[A-Za-z0-9][A-Za-z0-9_-]{0,199}$/; +const SETTINGS_BYTES = 256 * 1024; +const ENVIRON_BYTES = 1024 * 1024; + +export interface LettaSession { + reported: string; + /** null when the report names a conversation without its agent */ + agent: string | null; + conversation: string; +} + +/** The conversation herdr's Letta integration reported for the pane, or null for none or another shape. */ +export function lettaSessionOf(pane: Pick): LettaSession | null { + const session = pane.agent_session; + if (session?.agent !== "letta" || session.source !== "herdr:letta" || session.kind !== "id" || typeof session.value !== "string") return null; + const reported = session.value; + if (reported.startsWith("default:")) { + const agent = reported.slice("default:".length); + return ID.test(agent) ? { reported, agent, conversation: "default" } : null; + } + return ID.test(reported) && reported !== "default" ? { reported, agent: null, conversation: reported } : null; +} + +type Env = Record; + +/** Letta's config folder for an environment: `$LETTA_CONFIG_DIR` when absolute, else `$HOME/.letta`; null without either. */ +export function lettaConfigDir(env: Env): string | null { + const override = env["LETTA_CONFIG_DIR"]?.trim(); + if (override) return isAbsolute(override) ? normalize(override) : null; + return env["HOME"] && isAbsolute(env["HOME"]) ? join(env["HOME"], ".letta") : null; +} + +/** Where Letta writes transcripts: `$LETTA_TRANSCRIPT_ROOT`, else `transcripts` in its config folder. */ +export function lettaTranscriptRoot(env: Env): string | null { + const root = env["LETTA_TRANSCRIPT_ROOT"]?.trim(); + if (root) return isAbsolute(root) ? normalize(root) : null; + const config = lettaConfigDir(env); + return config === null ? null : join(config, "transcripts"); +} + +export const lettaTranscriptPath = (root: string, agent: string, conversation: string): string => + join(root, agent, conversation, "transcript.jsonl"); + +/** A NUL-separated environment, as /proc//environ holds it. */ +export function parseEnviron(text: string): Env { + const env: Env = {}; + for (const entry of text.split("\0")) { + const at = entry.indexOf("="); + if (at > 0) env[entry.slice(0, at)] = entry.slice(at + 1); + } + return env; +} + +function readBounded(path: string, bytes: number): string | null { + let fd: number; + try { fd = openSync(path, "r"); } catch { return null; } + try { + const size = fstatSync(fd).size; + if (size > bytes) return null; + const buffer = Buffer.alloc(size); + return buffer.subarray(0, readSync(fd, buffer, 0, size, 0)).toString("utf8"); + } catch { return null; } finally { closeSync(fd); } +} + +/** A process's environment, or null where it cannot be read. */ +function processEnvironment(pid: number): Env | null { + if (process.platform === "linux") { + try { + const bytes = readFileSync(`/proc/${pid}/environ`); + return bytes.length > ENVIRON_BYTES ? null : parseEnviron(bytes.toString("utf8")); + } catch { return null; } + } + if (process.platform === "darwin") { + const text = darwinProcess.environment(pid); + return text === null ? null : parseEnviron(text); + } + return null; +} + +/** + * The conversations a Letta settings file says it opened last, one per server; null when the text + * is not a settings object whose every session reads as Letta writes one. + */ +export function lettaOpenedSessions(text: string): { agent: string; conversation: string }[] | null { + let parsed: unknown; + try { parsed = JSON.parse(text); } catch { return null; } + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return null; + const servers = (parsed as Record)["sessionsByServer"]; + if (servers === undefined) return []; + if (servers === null || typeof servers !== "object" || Array.isArray(servers)) return null; + const sessions: { agent: string; conversation: string }[] = []; + for (const value of Object.values(servers as Record)) { + const ref = value !== null && typeof value === "object" ? value as Record : {}; + if (typeof ref["agentId"] !== "string" || typeof ref["conversationId"] !== "string") return null; + sessions.push({ agent: ref["agentId"], conversation: ref["conversationId"] }); + } + return sessions; +} + +/** + * The agent and conversation the report and Letta's record both name, or null. The agent comes + * from the record when the report names only a conversation (`local-conv-` repeats across + * agents), and every record of that agent must name the reported conversation. + */ +export function lettaBinding(session: LettaSession, opened: readonly { agent: string; conversation: string }[]): { agent: string; conversation: string } | null { + const named = opened.filter((ref) => ref.conversation === session.conversation && (session.agent === null || ref.agent === session.agent)); + const agents = new Set(named.map((ref) => ref.agent)); + if (agents.size !== 1) return null; + const agent = [...agents][0]!; + if (!ID.test(agent) || opened.some((ref) => ref.agent === agent && ref.conversation !== session.conversation)) return null; + return { agent, conversation: session.conversation }; +} + +/** A letta process: the `letta` executable, or a runtime running Letta's `letta` / `letta.js` entry. */ +export function isLettaProcess(argv: readonly string[]): boolean { + const name = (value: string | undefined): string => basename((value ?? "").replace(/\\/g, "/")).replace(/\.(exe|cmd)$/i, ""); + if (name(argv[0]) === "letta") return true; + return ["node", "bun"].includes(name(argv[0])) && ["letta", "letta.js"].includes(name(argv[1])); +} + +/** `startedAt`: no later than the process started (processStartLowerBound) */ +export interface LettaProcess { pid: number; cwd: string; startedAt: number; env: Env } + +/** Letta processes, by pid and start, seen in a folder other than the one they started in. */ +const moved = new Set(); + +/** + * Whether a letta process is in the folder it started in, and was in it every time it was looked at. + * Letta's SetWorkingDirectory and EnterWorktree tools move a running letta into another project without + * recording its conversation there, so that project's record is no record of it. The start folder is + * the `PWD` its shell gave it, which its environment keeps whatever it does later; without one, or + * once seen elsewhere, it is never bound again. + */ +export function lettaInStartFolder(owner: LettaProcess): boolean { + const key = `${owner.pid}:${owner.startedAt}`; + if (moved.has(key)) return false; + const start = owner.env["PWD"]; + const here = canonical(owner.cwd); + if (start === undefined || !isAbsolute(start) || here === null) return false; + if (canonical(start) === here) return true; + moved.add(key); + if (moved.size > 256) moved.delete(moved.values().next().value!); + return false; +} + +/** The pane's one letta process, or null for none, several, or one whose start or environment cannot be read. */ +export async function paneLettaProcess(paneId: string, known?: ProcessInfo | null): Promise { + const info = known === undefined ? await paneProcessInfo(paneId).catch(() => null) : known; + const found = (info?.foreground_processes ?? []).filter((process) => + Number.isSafeInteger(process.pid) && process.pid > 1 && Array.isArray(process.argv) && isLettaProcess(process.argv.map(String))); + if (found.length !== 1) return null; + const { pid, cwd } = found[0]!; + if (typeof cwd !== "string" || !isAbsolute(cwd)) return null; + const startedAt = processStartLowerBound(pid); + const env = processEnvironment(pid); + return startedAt === null || env === null ? null : { pid, cwd, startedAt, env }; +} + +export type LettaTranscript = { path: string; session: string; agent: string; conversation: string }; + +/** + * The transcript a Letta pane shows (see the top of this file), or null when it is not proven, or + * when another pane reports the same conversation. The file may not exist yet. + */ +export function lettaTranscriptFor(pane: HerdrPane, panes: readonly HerdrPane[], owner: LettaProcess): LettaTranscript | null { + const session = lettaSessionOf(pane); + if (session === null) return null; + if (panes.some((other) => other.pane_id !== pane.pane_id && lettaSessionOf(other)?.reported === session.reported)) return null; + const text = recordOlderThan(join(owner.cwd, ".letta", "settings.local.json"), owner.startedAt); + const opened = text === null ? null : lettaOpenedSessions(text); + const bound = opened === null ? null : lettaBinding(session, opened); + const root = lettaTranscriptRoot(owner.env); + if (bound === null || root === null) return null; + return { path: lettaTranscriptPath(root, bound.agent, bound.conversation), session: session.reported, ...bound }; +} + +/** + * Two seconds: a file system that keeps whole seconds (HFS+) or two (FAT) rounds a write after the + * start down to before it by less than this. + */ +const MTIME_SLACK_MS = 2000; + +/** + * The text of Letta's record when it is a regular file Letta can still rewrite (it writes in place and + * goes on after a failed write, which would leave an old record standing for a new conversation), last + * written at least MTIME_SLACK_MS before `startedAt`, and unchanged while it was read: its time and its + * words come from one descriptor, checked again after the read. Null otherwise. + */ +function recordOlderThan(path: string, startedAt: number): string | null { + let fd: number; + try { + if (!lstatSync(path).isFile()) return null; + accessSync(path, constants.W_OK); + fd = openSync(path, "r"); + } catch { return null; } + try { + const before = fstatSync(fd); + if (before.mtimeMs >= startedAt - MTIME_SLACK_MS || before.size > SETTINGS_BYTES) return null; + const buffer = Buffer.alloc(before.size); + const got = readSync(fd, buffer, 0, before.size, 0); + const after = fstatSync(fd); + if (got !== before.size || after.mtimeMs !== before.mtimeMs || after.size !== before.size) return null; + return buffer.toString("utf8"); + } catch { return null; } finally { closeSync(fd); } +} + +/** + * Whether `path` (lettaTranscriptPath under `root`) is a regular file reached through no link below + * the root: false for a missing file, a link, or one that cannot be checked. + */ +export function lettaTranscriptHeld(path: string, root: string): boolean { + try { + if (!lstatSync(path).isFile()) return false; + return realpathSync(path) === join(realpathSync(root), path.slice(root.length)); + } catch { return false; } +} + +const canonical = (path: string): string | null => { try { return realpathSync(path); } catch { return null; } }; + +/** + * Whether Letta's project record can speak for the pane's process at all: not in its home folder, + * where Letta with LETTA_CONFIG_DIR writes no project record (so an old one stays), and not while + * another pane runs Letta in the same folder, which writes the same record. + */ +export function lettaRecordOwned(pane: HerdrPane, panes: readonly HerdrPane[], owner: LettaProcess): boolean { + const folder = canonical(owner.cwd); + const home = owner.env["HOME"] === undefined ? null : canonical(owner.env["HOME"]); + if (folder === null || home === null || home === folder) return false; + return !panes.some((other) => other.pane_id !== pane.pane_id && (other.agent === "letta" || other.agent_session?.agent === "letta") + && [other.foreground_cwd, other.cwd].some((cwd) => typeof cwd === "string" && cwd !== "" && canonical(cwd) === folder)); +} + +/** Letta's local backend store for an environment: `$LETTA_LOCAL_BACKEND_DIR`, else `lc-local-backend` in its config folder. */ +export function lettaLocalStore(env: Env): string | null { + const dir = env["LETTA_LOCAL_BACKEND_DIR"]; + if (dir !== undefined) return isAbsolute(dir) ? normalize(dir) : null; + const config = lettaConfigDir(env); + return config === null ? null : join(config, "lc-local-backend"); +} + +/** + * Whether a local conversation (`local-conv-`) is the agent's own in the process's store, where + * Letta keys it by its id alone (`conversation:`, one folder each, never reused across agents) and + * names its agent. A cloud id (`conv-`) or `default` (keyed with its agent) needs no such check. + */ +export function lettaLocalOwner(store: string, agent: string, conversation: string): boolean { + if (!conversation.startsWith("local-conv-")) return true; + const record = join(store, "conversations", Buffer.from(`conversation:${conversation}`).toString("base64url"), "conversation.json"); + const text = readBounded(record, SETTINGS_BYTES); + if (text === null) return false; + try { + const parsed: unknown = JSON.parse(text); + return parsed !== null && typeof parsed === "object" && (parsed as Record)["id"] === conversation + && (parsed as Record)["agent_id"] === agent; + } catch { return false; } +} + +/** The transcript a Letta pane shows, written and held as lettaTranscriptHeld says, or null. */ +export async function resolveLettaTranscript(pane: HerdrPane, panes: readonly HerdrPane[], known?: ProcessInfo | null): Promise { + const owner = await paneLettaProcess(pane.pane_id, known); + if (owner === null || !lettaInStartFolder(owner) || !lettaRecordOwned(pane, panes, owner)) return null; + const found = lettaTranscriptFor(pane, panes, owner); + const store = lettaLocalStore(owner.env); + const root = lettaTranscriptRoot(owner.env); + if (found === null || store === null || root === null || !lettaLocalOwner(store, found.agent, found.conversation)) return null; + return lettaTranscriptHeld(found.path, root) ? found : null; +} diff --git a/server/letta.test.ts b/server/letta.test.ts new file mode 100644 index 0000000000..c87faa6d5d --- /dev/null +++ b/server/letta.test.ts @@ -0,0 +1,328 @@ +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { appendFileSync, chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, symlinkSync, utimesSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { forgetTranscriptState, paneTranscript, transcriptPage, transcriptToolOutput } from "./conversation.ts"; +import { ConversationUnavailable } from "./conversation-shared.ts"; +import * as herdr from "./herdr/client.ts"; +import { LettaUnavailable, lettaToolOutput, parseLettaTranscript } from "./letta.ts"; +import { isLettaProcess, type LettaProcess, lettaBinding, lettaInStartFolder, lettaLocalOwner, lettaRecordOwned, lettaOpenedSessions, lettaSessionOf, lettaTranscriptFor, lettaTranscriptHeld, lettaTranscriptPath, parseEnviron } from "./letta-store.ts"; +import type { HerdrPane, SessionSnapshot } from "../shared/protocol.ts"; + +// Recorded from letta-code 0.34.27 in an isolated herdr session (local backend, herdr's Letta +// integration installed): two turns in the default conversation, then `/new` and two more. +const fixture = (name: string): string => readFileSync(join(import.meta.dir, "fixtures", "letta", name), "utf8"); +const defaultLog = fixture("transcript-default-0.34.27.jsonl"); +const newLog = fixture("transcript-conversation-0.34.27.jsonl"); +const settingsAfterNew = fixture("settings.local-0.34.27.json"); +const localConversation = fixture("local-conversation-0.34.27.json"); +const AGENT = "agent-local-97f4bceb-2f0d-4a16-8b9f-abb2f3d4a72c"; + +const pane = (id: string, value: string | null, cwd: string): HerdrPane => ({ + pane_id: id, workspace_id: "w1", tab_id: "w1:t1", focused: false, cwd, foreground_cwd: cwd, agent: "letta", agent_status: "idle", + ...(value === null ? {} : { agent_session: { agent: "letta", kind: "id", source: "herdr:letta", value } }), +} as HerdrPane); + +let root: string; +beforeEach(() => { root = mkdtempSync(join(tmpdir(), "letta-")); }); +afterEach(() => { rmSync(root, { recursive: true, force: true }); forgetTranscriptState(); }); + +describe("Letta transcript", () => { + test("reads the prompts, tool calls with their results and answers Letta 0.34.27 recorded", () => { + const turns = parseLettaTranscript(defaultLog); + expect(turns.map((turn) => turn.role)).toEqual(["user", "assistant", "user", "assistant"]); + expect(turns[0]).toEqual({ role: "user", ts: "2026-10-29T01:53:05.293Z", parts: [{ kind: "text", text: "Use your shell tool to run: echo fixture-ok && ls. Then reply in one short sentence." }] }); + const [bash, answer] = turns[1]!.parts; + expect(bash).toMatchObject({ kind: "tool", name: "Bash", input: '{"command":"echo fixture-ok && ls","description":"Run test command to check fixture and list files"}', output: "fixture-ok\n" }); + expect(bash).not.toHaveProperty("error"); + expect(answer).toEqual({ kind: "text", text: "I have successfully run the command and verified the environment." }); + expect(turns[3]!.parts[0]).toMatchObject({ kind: "tool", name: "Read", output: "1\talpha\n2\tbeta\n3\t" }); + expect(turns[3]!.end_ts).toBe("2026-10-29T01:55:05.021Z"); + }); + + test("a call Letta recorded as failed is marked so", () => { + const turns = parseLettaTranscript(newLog); + expect(turns.map((turn) => turn.role)).toEqual(["user", "assistant", "user", "assistant"]); + expect(turns[3]!.parts[0]).toMatchObject({ kind: "tool", name: "Bash", error: true, output: "Exit code: 1\ncat: missing-file.txt: No such file or directory\n" }); + }); + + test("reasoning and errors keep their own kinds", () => { + const line = (entry: object): string => `${JSON.stringify({ captured_at: "2026-10-29T01:00:00.000Z", source_line_id: "x", ...entry })}\n`; + const turns = parseLettaTranscript(line({ kind: "user", text: "hi" }) + line({ kind: "reasoning", text: "thinking" }) + line({ kind: "error", text: "rate limited" })); + expect(turns[1]!.parts).toEqual([{ kind: "thinking", text: "thinking" }, { kind: "notice", text: "rate limited", source: "error" }]); + }); + + test("a transcript in a shape the reader does not know is refused, so the pane uses Terminal", () => { + expect(() => parseLettaTranscript(`${JSON.stringify({ kind: "summary", text: "x", captured_at: "t" })}\n`)).toThrow(LettaUnavailable); + expect(() => parseLettaTranscript(`{"kind":"user"\n${defaultLog}`)).toThrow(LettaUnavailable); + // the last line can be one Letta is still writing + expect(parseLettaTranscript(`${defaultLog}{"kind":"user","te`)).toHaveLength(4); + const path = join(root, "transcript.jsonl"); + writeFileSync(path, `${defaultLog}${JSON.stringify({ kind: "tool_call", name: 3, captured_at: "t" })}\n`); + expect(() => transcriptPage("letta-transcript", path)).toThrow(ConversationUnavailable); + }); + + test("pages at prompts and follows appends as a cold read does", () => { + const path = join(root, "transcript.jsonl"); + writeFileSync(path, defaultLog); + const first = transcriptPage("letta-transcript", path); + expect(first.turns).toHaveLength(4); + expect(first.metadata).toEqual({ model: null, reasoning_effort: null }); + appendFileSync(path, newLog); + const live = transcriptPage("letta-transcript", path); + expect(live.turns).toHaveLength(8); + expect(live.history_id).toBe(first.history_id); + forgetTranscriptState(); + expect(transcriptPage("letta-transcript", path).turns).toEqual(live.turns); + }); + + test("the whole output of a call is read by the ref its part carries", () => { + const path = join(root, "transcript.jsonl"); + const long = "y".repeat(200_000); + writeFileSync(path, `${JSON.stringify({ kind: "user", text: "go", captured_at: "2026-10-29T01:00:00.000Z", source_line_id: "user-1" })}\n${JSON.stringify({ kind: "tool_call", name: "Bash", argsText: "{}", resultText: long, resultOk: true, captured_at: "2026-10-29T01:00:00.000Z", source_line_id: "Bash-1-1" })}\n`); + const tool = transcriptPage("letta-transcript", path).turns[1]!.parts[0]!; + expect(tool).toMatchObject({ kind: "tool", output_ref: "Bash-1-1", output_size: long.length }); + expect(lettaToolOutput(path, "Bash-1-1")).toBe(long); + expect(transcriptToolOutput("letta-transcript", path, "Bash-1-1")?.startsWith("yyy")).toBe(true); + expect(lettaToolOutput(path, "Bash-9-9")).toBeNull(); + }); + + test("a field the reader does not know falls back, and a ref the output route refuses is not offered", () => { + const entry = (fields: object): string => `${JSON.stringify({ kind: "tool_call", name: "Bash", argsText: "{}", captured_at: "2026-10-29T01:00:00.000Z", ...fields })}\n`; + expect(() => parseLettaTranscript(entry({ result: { text: "must not drop" } }))).toThrow(LettaUnavailable); + const [part] = parseLettaTranscript(entry({ resultText: "z".repeat(10_000), source_line_id: "x".repeat(154) }))[0]!.parts; + expect(part).toMatchObject({ kind: "tool" }); + expect(part).not.toHaveProperty("output_ref"); + }); +}); + +describe("Letta conversation binding", () => { + const opened = lettaOpenedSessions(settingsAfterNew)!; + /** A project whose Letta record is `settings`, written `at`, and the pane's letta process started `started`. */ + const setup = (settings: string | null, at: number, started: number, env: Record = {}): LettaProcess => { + const cwd = join(root, "proj"); + mkdirSync(join(cwd, ".letta"), { recursive: true }); + if (settings !== null) { + writeFileSync(join(cwd, ".letta", "settings.local.json"), settings); + utimesSync(join(cwd, ".letta", "settings.local.json"), at / 1000, at / 1000); + } + return { pid: 4242, cwd, startedAt: started, env: { HOME: root, ...env } }; + }; + const now = Date.now(); + + test("reads herdr's report, Letta's record and a letta process as they were recorded", () => { + expect(lettaSessionOf(pane("w1:p1", `default:${AGENT}`, root))).toEqual({ reported: `default:${AGENT}`, agent: AGENT, conversation: "default" }); + expect(lettaSessionOf(pane("w1:p1", "local-conv-1", root))).toEqual({ reported: "local-conv-1", agent: null, conversation: "local-conv-1" }); + expect(opened).toEqual([{ agent: AGENT, conversation: "local-conv-1" }]); + expect(isLettaProcess(["node", "/home/user/.npm-global/bin/letta", "-m", "openai-compatible/gemini-3.1-flash-lite"])).toBe(true); + expect(isLettaProcess(["letta"])).toBe(true); + expect(isLettaProcess(["node", "/srv/app.js"])).toBe(false); + expect(parseEnviron("HOME=/h\0LETTA_CONFIG_DIR=/c=d\0")).toEqual({ HOME: "/h", LETTA_CONFIG_DIR: "/c=d" }); + }); + + test("binds the conversation both name, taking its agent from Letta's record", () => { + expect(lettaBinding(lettaSessionOf(pane("w1:p1", "local-conv-1", root))!, opened)).toEqual({ agent: AGENT, conversation: "local-conv-1" }); + }); + + test("a report /new left behind binds nothing: herdr kept default while Letta moved to local-conv-1", () => { + expect(lettaBinding(lettaSessionOf(pane("w1:p1", `default:${AGENT}`, root))!, opened)).toBeNull(); + }); + + test("a conversation id two agents' records name, or no record names, binds nothing", () => { + const session = lettaSessionOf(pane("w1:p1", "local-conv-1", root))!; + expect(lettaBinding(session, [...opened, { agent: "agent-local-other", conversation: "local-conv-1" }])).toBeNull(); + expect(lettaBinding(session, [])).toBeNull(); + expect(lettaBinding(session, [{ agent: AGENT, conversation: "local-conv-2" }])).toBeNull(); + }); + + test("reports of another source, agent or shape are not Letta's", () => { + const base = pane("w1:p1", "local-conv-1", root); + for (const agent_session of [ + { agent: "letta", kind: "id", source: "herdr:opencode", value: "local-conv-1" }, + { agent: "claude", kind: "id", source: "herdr:letta", value: "local-conv-1" }, + { agent: "letta", kind: "path", source: "herdr:letta", value: "local-conv-1" }, + { agent: "letta", kind: "id", source: "herdr:letta", value: "../../etc" }, + { agent: "letta", kind: "id", source: "herdr:letta", value: ".." }, + { agent: "letta", kind: "id", source: "herdr:letta", value: "default:.." }, + { agent: "letta", kind: "id", source: "herdr:letta", value: "default:" }, + { agent: "letta", kind: "id", source: "herdr:letta", value: "default" }, + ]) expect(lettaSessionOf({ ...base, agent_session } as HerdrPane)).toBeNull(); + expect(lettaSessionOf(pane("w1:p1", null, root))).toBeNull(); + }); + + test("Letta's record speaks for the pane's process only when it is older than that process", () => { + const reported = (cwd: string) => pane("w1:p1", "local-conv-1", cwd); + const path = lettaTranscriptPath(join(root, ".letta", "transcripts"), AGENT, "local-conv-1"); + // the process resumed the conversation the record names, so it wrote nothing + let owner = setup(settingsAfterNew, now - 60_000, now); + expect(lettaTranscriptFor(reported(owner.cwd), [reported(owner.cwd)], owner)).toEqual({ path, session: "local-conv-1", agent: AGENT, conversation: "local-conv-1" }); + // any write since: its own start on another conversation, its /new, or another letta + // a file system that keeps whole seconds (HFS+) or two (FAT) puts a write just after the start up to two seconds before it + for (const at of [now - 1_999, now - 999, now, now + 1, now + 2_000, now + 600_000]) { + owner = setup(settingsAfterNew, at, now); + expect(lettaTranscriptFor(reported(owner.cwd), [reported(owner.cwd)], owner)).toBeNull(); + } + }); + + test("a record Letta cannot rewrite binds nothing: its switch would fail and leave the old one standing", () => { + const owner = setup(settingsAfterNew, now - 60_000, now); + const target = pane("w1:p1", "local-conv-1", owner.cwd); + expect(lettaTranscriptFor(target, [target], owner)).not.toBeNull(); + chmodSync(join(owner.cwd, ".letta", "settings.local.json"), 0o444); + expect(lettaTranscriptFor(target, [target], owner)).toBeNull(); + }); + + test("a letta process is bound only in the folder it started in, and never again once seen elsewhere", () => { + const start = join(root, "project1"); + const other = join(root, "project2"); + mkdirSync(start); mkdirSync(other); + symlinkSync(start, join(root, "link1")); + const at = (pid: number, cwd: string, env: Record): LettaProcess => ({ pid, cwd, startedAt: now, env }); + expect(lettaInStartFolder(at(101, start, { PWD: start }))).toBe(true); + expect(lettaInStartFolder(at(102, start, { PWD: join(root, "link1") }))).toBe(true); + // no start folder to compare: its shell set no PWD + expect(lettaInStartFolder(at(103, start, {}))).toBe(false); + expect(lettaInStartFolder(at(104, start, { PWD: "relative" }))).toBe(false); + // SetWorkingDirectory moved it into project2; a later move back does not make project1's record its own again + expect(lettaInStartFolder(at(105, other, { PWD: start }))).toBe(false); + expect(lettaInStartFolder(at(105, start, { PWD: start }))).toBe(false); + }); + + test("a closed peer of the same agent that reopened the pane's old conversation binds nothing", () => { + // A started on local-conv-1 (herdr keeps that report), switched to local-conv-2; B opened local-conv-1 and closed. + // Every write was A's own or B's after A started, and the record names local-conv-1 again. + const owner = setup(JSON.stringify({ sessionsByServer: { local: { agentId: AGENT, conversationId: "local-conv-1" } } }), now + 4_000, now); + const store = join(root, ".letta", "lc-local-backend"); + const folder = join(store, "conversations", Buffer.from("conversation:local-conv-1").toString("base64url")); + mkdirSync(folder, { recursive: true }); + writeFileSync(join(folder, "conversation.json"), localConversation); + expect(lettaLocalOwner(store, AGENT, "local-conv-1")).toBe(true); + const a = pane("w1:p1", "local-conv-1", owner.cwd); + expect(lettaRecordOwned(a, [a], owner)).toBe(true); + expect(lettaTranscriptFor(a, [a], owner)).toBeNull(); + }); + + test("a missing, malformed or partly unreadable project record binds nothing, whatever the global settings say", () => { + mkdirSync(join(root, ".letta"), { recursive: true }); + writeFileSync(join(root, ".letta", "settings.json"), settingsAfterNew); + for (const settings of [null, "{", "[]", JSON.stringify({ sessionsByServer: { a: { agentId: AGENT, conversationId: "local-conv-1" }, b: { agentId: 3 } } })]) { + rmSync(join(root, "proj"), { recursive: true, force: true }); + const owner = setup(settings, now - 60_000, now); + expect(lettaTranscriptFor(pane("w1:p1", "local-conv-1", owner.cwd), [], owner)).toBeNull(); + } + }); + + test("the transcript is where the pane's process keeps it, never where this server would", () => { + const own = setup(settingsAfterNew, now - 60_000, now, { LETTA_TRANSCRIPT_ROOT: join(root, "elsewhere") }); + const target = pane("w1:p1", "local-conv-1", own.cwd); + expect(lettaTranscriptFor(target, [target], own)?.path).toBe(lettaTranscriptPath(join(root, "elsewhere"), AGENT, "local-conv-1")); + expect(lettaTranscriptFor(target, [target], { ...own, env: { HOME: root, LETTA_CONFIG_DIR: join(root, "cfg") } })?.path).toBe(lettaTranscriptPath(join(root, "cfg", "transcripts"), AGENT, "local-conv-1")); + expect(lettaTranscriptFor(target, [target], { ...own, env: { LETTA_TRANSCRIPT_ROOT: "relative" } })).toBeNull(); + expect(lettaTranscriptFor(target, [target], { ...own, env: {} })).toBeNull(); + expect(lettaTranscriptFor(target, [target, pane("w2:p1", "local-conv-1", own.cwd)], own)).toBeNull(); + }); + + test("the record speaks for nobody in the home folder, or while another pane runs Letta in the same folder", () => { + const owner = setup(settingsAfterNew, now - 60_000, now); + const target = pane("w1:p1", "local-conv-1", owner.cwd); + expect(lettaRecordOwned(target, [target], owner)).toBe(true); + // Letta with LETTA_CONFIG_DIR never rewrites ~/.letta/settings.local.json: an old one there is no record of this process + expect(lettaRecordOwned(target, [target], { ...owner, cwd: root, env: { HOME: root, LETTA_CONFIG_DIR: join(root, "cfg") } })).toBe(false); + expect(lettaRecordOwned(target, [target], { ...owner, cwd: root })).toBe(false); + expect(lettaRecordOwned(target, [target], { ...owner, env: {} })).toBe(false); + // pane B reported default:agent-b, then /new wrote its own local-conv-1 into the shared record + symlinkSync(owner.cwd, join(root, "link")); + const peer = { ...pane("w2:p1", "default:agent-local-b", join(root, "link")), cwd: join(root, "elsewhere") } as HerdrPane; + expect(lettaRecordOwned(target, [target, peer], owner)).toBe(false); + expect(lettaRecordOwned(target, [target, { ...peer, agent: "letta", agent_session: undefined } as HerdrPane], owner)).toBe(false); + expect(lettaRecordOwned(target, [target, { ...peer, agent: "claude", agent_session: undefined } as HerdrPane], owner)).toBe(true); + expect(lettaRecordOwned(target, [target, pane("w3:p1", "conv-x", root)], owner)).toBe(true); + }); + + test("a local conversation binds only to the agent the process's own store names for it", () => { + const store = join(root, "store"); + const folder = join(store, "conversations", Buffer.from("conversation:local-conv-1").toString("base64url")); + expect(lettaLocalOwner(store, AGENT, "local-conv-1")).toBe(false); + mkdirSync(folder, { recursive: true }); + writeFileSync(join(folder, "conversation.json"), localConversation); + expect(lettaLocalOwner(store, AGENT, "local-conv-1")).toBe(true); + // pane B's agent ran /new and left its own record naming local-conv-1: the store says whose it is + expect(lettaLocalOwner(store, "agent-local-b", "local-conv-1")).toBe(false); + writeFileSync(join(folder, "conversation.json"), "{"); + expect(lettaLocalOwner(store, AGENT, "local-conv-1")).toBe(false); + expect(lettaLocalOwner(store, AGENT, "conv-0d9d5c5e-3c3a-4a51-9d0c-1f5f2c1b7a11")).toBe(true); + expect(lettaLocalOwner(store, AGENT, "default")).toBe(true); + }); + + test("only a regular file at the path Letta derives is read, never one reached through a link", () => { + const transcripts = join(root, "transcripts"); + const path = lettaTranscriptPath(transcripts, AGENT, "local-conv-1"); + expect(lettaTranscriptHeld(path, transcripts)).toBe(false); + mkdirSync(join(path, ".."), { recursive: true }); + writeFileSync(join(root, "other.jsonl"), defaultLog); + symlinkSync(join(root, "other.jsonl"), path); + expect(lettaTranscriptHeld(path, transcripts)).toBe(false); + rmSync(path); + symlinkSync(join(root, "loop"), join(root, "loop")); + symlinkSync(join(root, "loop"), path); + expect(lettaTranscriptHeld(path, transcripts)).toBe(false); + rmSync(path); + writeFileSync(path, newLog); + expect(lettaTranscriptHeld(path, transcripts)).toBe(true); + rmSync(join(transcripts, AGENT), { recursive: true }); + mkdirSync(join(root, "real", "local-conv-1"), { recursive: true }); + writeFileSync(join(root, "real", "local-conv-1", "transcript.jsonl"), newLog); + symlinkSync(join(root, "real"), join(transcripts, AGENT)); + expect(lettaTranscriptHeld(path, transcripts)).toBe(false); + }); + + test("a bound pane reads the transcript its letta process keeps; without one, or unbound, it uses Terminal", async () => { + const project = join(root, "proj"); + mkdirSync(join(project, ".letta"), { recursive: true }); + writeFileSync(join(project, ".letta", "settings.local.json"), settingsAfterNew); + // written before the pane's letta started, as Letta leaves it when it resumes that conversation + utimesSync(join(project, ".letta", "settings.local.json"), (Date.now() - 60_000) / 1000, (Date.now() - 60_000) / 1000); + const config = join(root, "config"); + // a real process whose own environment names the store, as the pane's letta would + const child = Bun.spawn(["sleep", "300"], { env: { HOME: root, PWD: project, LETTA_CONFIG_DIR: config, PATH: process.env["PATH"] ?? "" }, cwd: project }); + // the same process after SetWorkingDirectory moved it from project1 into this project: its environment keeps project1 + mkdirSync(join(root, "project1")); + const movedIn = Bun.spawn(["sleep", "300"], { env: { HOME: root, PWD: join(root, "project1"), LETTA_CONFIG_DIR: config, PATH: process.env["PATH"] ?? "" }, cwd: project }); + let foreground = child.pid; + const info = spyOn(herdr, "paneProcessInfo").mockImplementation(async () => ({ + pane_id: "w1:p1", foreground_processes: [{ pid: foreground, argv: ["node", "/opt/letta/bin/letta"], cwd: project }], + }) as never); + const rpc = spyOn(herdr, "herdrRpc").mockImplementation(async () => ({}) as never); + const resolve = (target: HerdrPane) => paneTranscript(target.pane_id, undefined, { panes: [target] } as SessionSnapshot); + try { + const bound = pane("w1:p1", "local-conv-1", project); + const path = lettaTranscriptPath(join(config, "transcripts"), AGENT, "local-conv-1"); + // a conversation with no file here may have turns elsewhere: the terminal shows them + await expect(resolve(bound)).rejects.toThrow("no_session_path"); + mkdirSync(join(path, ".."), { recursive: true }); + writeFileSync(path, newLog); + // the store has no record that local-conv-1 is this agent's + await expect(resolve(bound)).rejects.toThrow("no_session_path"); + const record = join(config, "lc-local-backend", "conversations", Buffer.from("conversation:local-conv-1").toString("base64url")); + mkdirSync(record, { recursive: true }); + writeFileSync(join(record, "conversation.json"), localConversation); + expect(await resolve(bound)).toEqual({ source: "letta-transcript", path }); + await expect(resolve(pane("w1:p1", `default:${AGENT}`, project))).rejects.toThrow("no_session_path"); + await expect(resolve(pane("w1:p1", null, project))).rejects.toThrow("no_session_path"); + // another pane running Letta in the folder writes the same record + // project1 -> /new -> moved into this project, whose older record names the conversation it left + foreground = movedIn.pid; + await expect(resolve(bound)).rejects.toThrow("no_session_path"); + foreground = child.pid; + expect(await resolve(bound)).toEqual({ source: "letta-transcript", path }); + const peer = pane("w2:p1", "default:agent-local-b", project); + await expect(paneTranscript(bound.pane_id, undefined, { panes: [bound, peer] } as SessionSnapshot)).rejects.toThrow("no_session_path"); + } finally { + info.mockRestore(); + rpc.mockRestore(); + child.kill(); + movedIn.kill(); + await Promise.all([child.exited, movedIn.exited]); + } + }); +}); diff --git a/server/letta.ts b/server/letta.ts new file mode 100644 index 0000000000..64423bf5f1 --- /dev/null +++ b/server/letta.ts @@ -0,0 +1,116 @@ +/** + * Letta Code's conversation transcript, read into turns. The TUI appends each turn when it ends + * (`end_turn`) to `///transcript.jsonl`, on the cloud and the + * local backend alike (letta-code 0.34.27, `appendTranscriptDeltaJsonl`). A line is one entry: + * - `{ kind: "user" | "assistant" | "reasoning" | "error", text, captured_at }`; + * - `{ kind: "tool_call", name?, argsText?, resultText?, resultOk?, captured_at }`, the call and its + * result in one entry. + * Each also names a `source_line_id` (unique in the file) and may name a `source_message_id`. Every + * entry of one turn has the same `captured_at`. A turn that was cancelled or failed is not written, + * and a turn still running is written only when it ends. A whole line that does not parse or is in + * another shape throws LettaUnavailable: Chat then uses Terminal rather than show part of it. + */ +import { closeSync, openSync, readSync } from "node:fs"; +import type { ConversationPart, ConversationTurn } from "../shared/protocol.ts"; +import { trimOutput } from "./tool-output.ts"; +import { toolSummary } from "./transcript-records.ts"; + +type Row = Record; +type Tool = Extract; +const isRow = (value: unknown): value is Row => value !== null && typeof value === "object" && !Array.isArray(value); +const optional = (value: unknown, type: "string" | "boolean"): boolean => value === undefined || typeof value === type; + +export class LettaUnavailable extends Error { + constructor() { super("Letta transcript is in a shape the reader does not know"); this.name = "LettaUnavailable"; } +} + +/** The fields letta-code 0.34.27 writes, by kind: an entry with any other could carry words the reader would drop. */ +const FIELDS = new Set(["kind", "text", "captured_at", "source_line_id", "source_message_id"]); +const TOOL_FIELDS = new Set(["kind", "name", "argsText", "resultText", "resultOk", "captured_at", "source_line_id", "source_message_id"]); +/** The refs GET /api/pane/conversation/tool-output takes (TOOL_REF in conversation.ts). */ +const TOOL_REF = /^[A-Za-z0-9_:.-]{1,128}$/; + +/** Whether a line is an entry letta-code 0.34.27 writes, in the shape the reader takes. */ +function known(entry: Row): boolean { + if (!Object.keys(entry).every((key) => (entry.kind === "tool_call" ? TOOL_FIELDS : FIELDS).has(key))) return false; + if (typeof entry.captured_at !== "string" || !optional(entry.source_line_id, "string") || !optional(entry.source_message_id, "string")) return false; + if (entry.kind === "tool_call") { + return optional(entry.name, "string") && optional(entry.argsText, "string") && optional(entry.resultText, "string") && optional(entry.resultOk, "boolean"); + } + return (entry.kind === "user" || entry.kind === "assistant" || entry.kind === "reasoning" || entry.kind === "error") && typeof entry.text === "string"; +} + +/** The words of a user entry, or null when the line is none: a page starts at such a line. */ +export function lettaPrompt(entry: Row): string | null { + return entry.kind === "user" && typeof entry.text === "string" && entry.text.trim() !== "" ? entry.text : null; +} + +function iso(value: unknown): string | null { + if (typeof value !== "string") return null; + const time = Date.parse(value); + return Number.isFinite(time) ? new Date(time).toISOString() : null; +} + +export function parseLettaTranscript(text: string): ConversationTurn[] { + const turns: ConversationTurn[] = []; + const answer = (ts: string | null, part: ConversationPart): void => { + const last = turns.at(-1); + if (last?.role === "assistant") { last.parts.push(part); if (ts !== null) last.end_ts = ts; } + else turns.push({ role: "assistant", ts, ...(ts === null ? {} : { end_ts: ts }), parts: [part] }); + }; + const lines = text.split("\n"); + for (const [at, line] of lines.entries()) { + if (line === "") continue; + let entry: unknown; + // the last line can be one Letta is still writing + try { entry = JSON.parse(line); } catch { if (at < lines.length - 1) throw new LettaUnavailable(); continue; } + if (!isRow(entry) || !known(entry)) throw new LettaUnavailable(); + const ts = iso(entry.captured_at); + const prompt = lettaPrompt(entry); + if (prompt !== null) { turns.push({ role: "user", ts, parts: [{ kind: "text", text: prompt }] }); continue; } + const words = typeof entry.text === "string" ? entry.text : ""; + if (entry.kind === "assistant" && words.trim() !== "") answer(ts, { kind: "text", text: words }); + else if (entry.kind === "reasoning" && words.trim() !== "") answer(ts, { kind: "thinking", text: words }); + else if (entry.kind === "error" && words.trim() !== "") answer(ts, { kind: "notice", text: words, source: "error" }); + else if (entry.kind === "tool_call") { + const name = typeof entry.name === "string" && entry.name !== "" ? entry.name : "tool"; + const input = typeof entry.argsText === "string" ? entry.argsText : ""; + let args: Row = {}; + try { const parsed: unknown = JSON.parse(input); if (isRow(parsed)) args = parsed; } catch { /* summarized by name */ } + const part: Tool = { kind: "tool", name, summary: toolSummary(name, args), input, output: "" }; + const ref = typeof entry.source_line_id === "string" && TOOL_REF.test(entry.source_line_id) ? entry.source_line_id : ""; + trimOutput(part, typeof entry.resultText === "string" ? entry.resultText : "", ref); + // an entry without an id the output route takes has no ref to fetch the rest by + if (ref === "") { delete part.output_ref; delete part.output_size; } + if (entry.resultOk === false) part.error = true; + answer(ts, part); + } + } + return turns; +} + +/** + * The whole result of one call, by the `source_line_id` of its entry. ponytail: reads the + * transcript from the start on each request; read backwards from the end if they grow large. + */ +export function lettaToolOutput(path: string, ref: string): string | null { + const id = JSON.stringify(ref); + let fd: number; + try { fd = openSync(path, "r"); } catch { return null; } + try { + const chunk = Buffer.alloc(1024 * 1024); + let carry = Buffer.alloc(0); + for (let position = 0, got = 0; (got = readSync(fd, chunk, 0, chunk.length, position)) > 0; position += got) { + const bytes = Buffer.concat([carry, chunk.subarray(0, got)]); + const end = bytes.lastIndexOf(0x0a) + 1; + carry = Buffer.from(bytes.subarray(end)); + for (const line of bytes.subarray(0, end).toString("utf8").split("\n")) { + if (!line.includes(id) || !line.includes('"tool_call"')) continue; + let entry: unknown; + try { entry = JSON.parse(line); } catch { continue; } + if (isRow(entry) && entry.kind === "tool_call" && entry.source_line_id === ref) return typeof entry.resultText === "string" ? entry.resultText : ""; + } + } + return null; + } catch { return null; } finally { closeSync(fd); } +} diff --git a/server/transcript-turns.ts b/server/transcript-turns.ts index 5a78c8e895..193794e806 100644 --- a/server/transcript-turns.ts +++ b/server/transcript-turns.ts @@ -4,6 +4,7 @@ import type { TranscriptEntry } from "./claude-transcript.ts"; import { claudeImages, isCommandEntry, withoutPictureBytes } from "./claude-transcript.ts"; import { copilotRootRecord } from "./copilot.ts"; import { musePrompt } from "./muse.ts"; +import { lettaPrompt } from "./letta.ts"; /** * A Claude session's branch, followed as the file is scanned. Claude Code links each record to the @@ -92,6 +93,7 @@ const TURN_MARK: Record = { "grok-transcript": Buffer.from('"user_message_chunk"'), "agy-transcript": Buffer.from('"USER_INPUT"'), "muse-transcript": Buffer.from('"payload_type":"runtime.user_intent.accepted"'), + "letta-transcript": Buffer.from('"kind":"user"'), "codex-transcript": Buffer.from('"task_started"'), "claude-transcript": Buffer.from('"user"'), "omp-transcript": Buffer.from('"user"'), @@ -115,6 +117,7 @@ function opensTurn(source: StreamSource, line: string, blobs: string | null): bo && typeof entry.data?.content === "string" && !!entry.data.content.trim(); if (source === "codex-transcript") return entry.type === "event_msg" && entry.payload?.type === "task_started"; if (source === "muse-transcript") return musePrompt(entry as Record) !== null; + if (source === "letta-transcript") return lettaPrompt(entry as Record) !== null; if (source !== "claude-transcript") { const message = piMessage(entry); if (message?.role !== "user") return false; diff --git a/shared/protocol.ts b/shared/protocol.ts index 5b42f952e5..caa8a6aa75 100644 --- a/shared/protocol.ts +++ b/shared/protocol.ts @@ -518,7 +518,7 @@ export interface ConversationResponse { activity?: "thinking" | "working" | "idle"; /** Stable across appends; changes on transcript replacement or native context clear. */ history_id?: string; - source: "copilot-transcript" | "grok-transcript" | "agy-transcript" | "muse-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "cmd-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript" | "scrollback"; + source: "copilot-transcript" | "grok-transcript" | "agy-transcript" | "muse-transcript" | "letta-transcript" | "claude-transcript" | "omp-transcript" | "omo-transcript" | "gjc-transcript" | "pi-transcript" | "cmd-transcript" | "codex-transcript" | "devin-transcript" | "opencode-transcript" | "hermes-transcript" | "scrollback"; turns: ConversationTurn[]; metadata?: ConversationMetadata; /**