Something went wrong. Try again.
ACP support (in agent-shell within Emacs) for Letta Code
Something went wrong. Try again.
39 kB · 963 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964#!/usr/bin/env nodeimport { AgentSideConnection, ndJsonStream, type Agent, type InitializeRequest, type InitializeResponse, type AuthenticateRequest, type AuthenticateResponse, type NewSessionRequest, type NewSessionResponse, type PromptRequest, type PromptResponse, type CancelNotification, type LoadSessionRequest, type LoadSessionResponse, type ListSessionsRequest, type ListSessionsResponse, type SetSessionModeRequest, type SetSessionModeResponse, type RequestPermissionRequest, type RequestPermissionResponse, type ContentBlock, type Diff,} from "@agentclientprotocol/sdk";import { Readable, Writable } from "node:stream";import { appendFileSync } from "node:fs";import { LettaProcess } from "./letta-process.js";import { wireToSessionUpdate, decorateMemoryToolCall, decorateFileEditToolCall, descriptiveTitle, normalizeToolInput, stripReadLineNumbers,} from "./mapping.js";import { buildPermissionRequestToolCall } from "./permission-request.js";import { UpdateGate } from "./update-gate.js";import { QuiescenceTracker, type QuiescenceAction, type ResolveReason } from "./quiescence.js";import { SessionPermissionGrants } from "./permission-policy.js";import { completionContentForOrphan, shouldKeepSuccessfulToolReturnContent, shouldSuppressToolCallRow,} from "./tool-display.js";import type { ResultWire, ErrorWire } from "./wire.js";
const PROTOCOL_VERSION = 1 as const;
/** * Debounce window (ms) after a queue-draining `result` before declaring the * turn quiescent - catches a continuation item that enqueues just after the * result. Only applies to turns that showed queue activity; plain turns resolve * instantly. 0 disables the debounce (resolve as soon as the queue reads idle). * Tune against a real capture via LETTA_ACP_TRACE. */const QUIESCENCE_DEBOUNCE_MS = envInt("LETTA_ACP_QUIESCENCE_MS", 400);/** * Hard cap (ms) on how long a single ACP turn may be held open waiting for the * queue to drain. A stuck/never-draining queue force-resolves at this bound so * a turn can never hang forever. 0 disables the cap (not recommended). */const MAX_HOLD_MS = envInt("LETTA_ACP_MAX_HOLD_MS", 600_000);/** * When set to a writable file path, every raw wire message is appended as a * JSONL line ({ts, msg}) for offline diagnosis of queue timing. No behavior * change; purely observational. */const TRACE_PATH = process.env.LETTA_ACP_TRACE?.trim() || null;
function envInt(name: string, fallback: number): number { const raw = process.env[name]; if (raw === undefined || raw === "") return fallback; const n = Number(raw); return Number.isFinite(n) && n >= 0 ? n : fallback;}
interface SessionState { id: string; proc: LettaProcess; pending?: { resolve: (r: PromptResponse) => void; reject: (e: Error) => void; }; /** * Tool-call ids we have emitted a `tool_call` notification for. Used to * dedupe - Letta streams the same tool_call_id across many wire deltas (one * per arg chunk), and we want exactly one ACP `tool_call` per real tool. */ seenToolCalls: Set<string>; /** * Diff content blocks emitted on the initial `tool_call` for memory tools. * Keyed by tool_call_id so we can splice them back into the final * `tool_call_update` (status=completed) content array - otherwise the tool * return text overwrites the diff and the user loses sight of what was * changed. */ memoryDiffs: Map<string, unknown>; /** Map tool_call_id -> tool name, so refinement updates can re-decorate by name. */ titlesById: Map<string, string>; /** * Map tool_call_id -> raw Letta tool name (e.g. "Read", "Edit"). titlesById * gets overwritten with descriptive titles ("Read ~/foo.ts") during * refinement, so we keep the raw name in a separate map for content * post-processing on completion (Read line-number stripping, Edit/Write * diff swap-in). */ toolNamesById: Map<string, string>; /** * Accumulated argument-string fragments per tool_call_id. Letta streams * tool arguments as many wire deltas (".arguments" carries a fragment of * the eventual JSON in each). We concatenate here and emit a single * refinement once the cumulative string parses, so the host sees one * `tool_call_update` with complete rawInput + memory diff content rather * than a flurry of partial updates. */ argsAccum: Map<string, string>; /** * Tool-call ids that have an emitted `tool_call` but no terminal * `tool_call_update` yet. Drained on `tool_return_message` (→ completed) or * on the terminating `result` message (→ failed, since the absence of a * return generally means the agent abandoned that call mid-turn, e.g. after * a tool error caused a retry with a fresh id). */ inflightTools: Set<string>; /** Tool-call ids represented by a richer ACP surface such as `plan`. */ suppressedToolCalls: Set<string>; /** * Tracks in-flight `session/update` notifications so we can flush them before * resolving a `session/prompt` (ending the turn). Prevents an update whose * write is still draining from landing after the prompt response, which ACP * hosts reject as "Out of turn". All `session/update` emissions for this * session MUST go through `gate.emit(...)`, never `conn.sessionUpdate(...)` * directly, or the ordering guarantee is lost. */ gate: UpdateGate; /** * Decides when the ACP turn is truly finished. Letta can stream continuation * runs (subagent completions, reflection) after the parent `result`; this * holds the prompt open until the turn queue drains instead of closing on the * first result. See quiescence.ts. */ quiescence: QuiescenceTracker; /** * Tools the user granted "Always allow" to this session. Subsequent * can_use_tool requests for a granted tool are auto-approved without * re-prompting (mirrors the interactive CLI). See permission-policy.ts. */ permissionGrants: SessionPermissionGrants; /** Debounce timer for the post-drain quiescence window. */ quiescenceTimer?: ReturnType<typeof setTimeout>; /** Safety cap timer that force-resolves a turn held too long. */ maxHoldTimer?: ReturnType<typeof setTimeout>;}
class LettaAcpAgent implements Agent { private sessions = new Map<string, SessionState>(); private clientCwd: string | null = null;
constructor(private conn: AgentSideConnection) {}
async initialize(_params: InitializeRequest): Promise<InitializeResponse> { return { protocolVersion: PROTOCOL_VERSION, agentCapabilities: { loadSession: false, promptCapabilities: { image: false, audio: false, embeddedContext: true, }, sessionCapabilities: { // v0: we accept the call but always return an empty list. // This silences agent-shell's startup probe ("Method not found: session/list"). // Real persistence (mapping ACP sessionIds to Letta agent/conversation // tuples across adapter restarts) is a v1 task. list: {}, }, }, authMethods: [ { id: "letta-cloud", name: "Letta Cloud", description: "Use Letta Cloud (set LETTA_API_KEY in adapter env).", }, { id: "letta-local", name: "Local backend", description: "Use the local Letta backend (run `letta connect <provider>` first).", }, ], }; }
async authenticate(_params: AuthenticateRequest): Promise<AuthenticateResponse> { // No-op: real auth lives in the `letta` CLI's own credential cache. // For cloud, LETTA_API_KEY must be present in the adapter's environment. return {}; }
async newSession(params: NewSessionRequest): Promise<NewSessionResponse> { const cwd = params.cwd ?? process.cwd(); this.clientCwd = cwd;
const meta = (params as any)._meta ?? {}; const agentId: string | undefined = meta.lettaAgentId ?? process.env.LETTA_AGENT_ID; const model: string | undefined = meta.lettaModel ?? process.env.LETTA_MODEL; const permissionMode: string | undefined = meta.lettaPermissionMode ?? process.env.LETTA_PERMISSION_MODE; const conversationId: string | undefined = meta.lettaConversationId ?? process.env.LETTA_CONVERSATION_ID; const reflectionTrigger: string | undefined = meta.lettaReflectionTrigger ?? process.env.LETTA_REFLECTION_TRIGGER; const reflectionStepCountRaw = meta.lettaReflectionStepCount ?? process.env.LETTA_REFLECTION_STEP_COUNT; const reflectionStepCount: number | undefined = reflectionStepCountRaw !== undefined && reflectionStepCountRaw !== null ? Number(reflectionStepCountRaw) : undefined; if ( reflectionStepCount !== undefined && (!Number.isFinite(reflectionStepCount) || reflectionStepCount <= 0) ) { throw rpcError( -32602, `lettaReflectionStepCount must be a positive integer, got ${reflectionStepCountRaw}`, ); }
const proc = new LettaProcess({ cwd, agentId, model, permissionMode, conversationId, newConversation: !!meta.newConversation, reflectionTrigger, reflectionStepCount, });
const init = await proc.start(); const sessionId = init.session_id ?? init.agent_id ?? `letta-${Date.now()}`;
const state: SessionState = { id: sessionId, proc, seenToolCalls: new Set(), inflightTools: new Set(), memoryDiffs: new Map(), titlesById: new Map(), toolNamesById: new Map(), argsAccum: new Map(), suppressedToolCalls: new Set(), gate: new UpdateGate((params) => this.conn.sessionUpdate(params as any)), quiescence: new QuiescenceTracker(), permissionGrants: new SessionPermissionGrants(), }; this.sessions.set(sessionId, state);
proc.onWire((msg) => this.dispatchWire(state, msg)); proc.on("exit", (info: { code: number | null; stderr: string }) => { this.clearTurnTimers(state); state.quiescence.markResolved(); const pending = state.pending; if (pending) { state.pending = undefined; if (info.code === 0) { pending.resolve({ stopReason: "end_turn" }); } else { pending.reject(new Error(`letta exited: ${info.stderr.trim() || `code ${info.code}`}`)); } } this.sessions.delete(sessionId); });
return { sessionId }; }
async loadSession(_params: LoadSessionRequest): Promise<LoadSessionResponse> { throw rpcError(-32601, "session/load not implemented in v0"); }
async listSessions(_params: ListSessionsRequest): Promise<ListSessionsResponse> { return { sessions: [] }; }
async prompt(params: PromptRequest): Promise<PromptResponse> { const state = this.sessions.get(params.sessionId); if (!state) throw rpcError(-32602, `unknown sessionId: ${params.sessionId}`); if (state.pending) throw rpcError(-32000, "another prompt is already in flight");
const text = extractText(params);
return new Promise<PromptResponse>((resolve, reject) => { state.pending = { resolve, reject }; // Begin the turn (arms the max-hold cap) before any wire arrives. this.applyQuiescence(state, state.quiescence.begin()); state.proc.sendUserPrompt(text); }); }
async cancel(params: CancelNotification): Promise<void> { const state = this.sessions.get(params.sessionId); if (!state) return; // Propagate the interrupt to Letta so the underlying run/subagent actually // stops - not just unlock the prompt. Then close the (possibly held-open) // ACP turn as cancelled. state.proc.cancel(); this.applyQuiescence(state, state.quiescence.onCancel()); }
async setSessionMode(_params: SetSessionModeRequest): Promise<SetSessionModeResponse> { // v0: we can't change Letta's --permission-mode mid-process. Accept the // request so ACP hosts (Zed, agent-shell) stop erroring; ignore the // requested mode. Mid-session change is a Phase 2 item requiring a Letta // control_request we haven't seen on the wire yet - at that point persist // the chosen mode and forward it. Today there's no reader, so no store. return {}; }
private dispatchWire(state: SessionState, msg: any): void { this.trace(msg);
// Queue lifecycle events drive the turn-hold decision: they tell us whether // continuation work (subagent completions, reflection) is still pending so // we keep the ACP prompt open instead of closing it on the first `result`. switch (msg.type) { case "queue_item_enqueued": this.applyQuiescence(state, state.quiescence.onEnqueued(Number(msg.queue_len))); return; case "queue_batch_dequeued": this.applyQuiescence(state, state.quiescence.onDequeued(Number(msg.queue_len_after))); return; case "queue_blocked": this.applyQuiescence( state, state.quiescence.onBlocked(String(msg.reason ?? ""), Number(msg.queue_len)), ); return; case "queue_cleared": this.applyQuiescence(state, state.quiescence.onCleared()); return; case "queue_item_dropped": // An item left the queue without processing; treat like a dequeue to 0 // floor so outstanding can't be stuck above reality. this.applyQuiescence(state, state.quiescence.onDequeued(Number(msg.queue_len))); return; }
if (msg.type === "result") { const subtype = (msg as any).subtype; this.completeInflightTools(state, { // result.subtype="success" means the agent's turn ended normally. // Any tool that streamed args but never returned was likely a // server-side tool whose `tool_return_message` didn't reach the // wire - surface as `completed` with a soft note, not `failed`. // result.subtype !== "success" (error / interrupted) → mark failed. treatOrphansAs: subtype === "success" ? "completed" : "failed", note: subtype === "success" ? "(no tool_return on wire - likely a server-side tool; agent may still have received the result)" : "(no tool_return - turn ended in error or was interrupted)", }); if (subtype === "success") { // Don't close the turn yet - let the quiescence tracker decide whether // more queued work is coming. It will resolve immediately for a plain // turn, or hold and resolve once the queue drains. this.applyQuiescence(state, state.quiescence.onSuccessResult()); } else { // interrupted / error results are terminal now (no queue can revive the // turn). Stop the tracker and resolve via the existing path. this.clearTurnTimers(state); state.quiescence.markResolved(); void this.resolvePrompt(state, msg as ResultWire).catch((err) => { process.stderr.write(`[letta-acp] resolvePrompt failed: ${err}\n`); }); } return; } if (msg.type === "error") { // Run-level error: tools in flight aren't going to complete. Mark them // failed with the actual run error message so the user can see WHY. const errText = String((msg as any).message ?? "Letta run error"); this.completeInflightTools(state, { treatOrphansAs: "failed", note: `Run error: ${errText.slice(0, 400)}${errText.length > 400 ? "…" : ""}`, }); this.clearTurnTimers(state); state.quiescence.markResolved(); void this.rejectPrompt(state, msg as ErrorWire).catch((err) => { process.stderr.write(`[letta-acp] rejectPrompt failed: ${err}\n`); }); return; } if (msg.type === "control_request") { this.handleControlRequest(state, msg).catch((err) => { process.stderr.write(`[letta-acp] control_request handler failed: ${err}\n`); }); return; }
// Accumulate streamed arg fragments so we can emit one refined // tool_call_update with full rawInput (+ memory diff) when the cumulative // JSON parses. Each Letta delta carries a fragment string. if (msg.type === "stream_event") { const ev: any = (msg as any).event ?? {}; const mt = ev.message_type; const tc = ev.tool_call; const id = tc?.tool_call_id; const frag = tc?.arguments; if ( (mt === "tool_call_message" || mt === "approval_request_message") && typeof id === "string" && typeof frag === "string" && frag.length > 0 ) { const accum = (state.argsAccum.get(id) ?? "") + frag; state.argsAccum.set(id, accum); try { const parsed = JSON.parse(accum); // Complete JSON. Emit one refined tool_call_update and stop // accumulating for this id. state.argsAccum.delete(id); const toolName = state.titlesById.get(id) ?? tc.name ?? ""; const base: any = { sessionUpdate: "tool_call_update", toolCallId: id, rawInput: normalizeToolInput(String(toolName), parsed), }; // Add a descriptive title (e.g. "Read ~/path/file.el") once we // have full args. agent-shell upgrades titles when a non-empty // new title arrives on tool_call_update. const nicer = descriptiveTitle(String(toolName), parsed); if (nicer) base.title = nicer; const refined = decorateFileEditToolCall( decorateMemoryToolCall(base, String(toolName), parsed), String(toolName), parsed, ); if (Array.isArray((refined as any).content)) { const diff = (refined as any).content.find( (c: any) => c?.type === "diff", ); if (diff) state.memoryDiffs.set(id, diff); } // For TodoWrite/UpdatePlan, also emit an ACP `plan` notification. // agent-shell has dedicated plan rendering (see agent-shell.el line // ~1697), which is far more useful than a generic tool_call row. const isPlanTool = String(toolName) === "TodoWrite" || String(toolName) === "UpdatePlan"; if (!isPlanTool) { state.gate.emit({ sessionId: state.id, update: refined } as any); } if (isPlanTool && Array.isArray((parsed as any).todos)) { const entries = ((parsed as any).todos as any[]).map((t) => ({ content: String(t?.content ?? t?.text ?? "(unnamed task)"), priority: String(t?.priority ?? "medium"), status: String(t?.status ?? "pending"), })); state.gate.emit({ sessionId: state.id, update: { sessionUpdate: "plan", entries }, } as any); } } catch { // not yet complete; keep accumulating } // Suppress per-fragment downstream mapping for this delta - the // initial tool_call already fired on the name-bearing delta. if (!tc.name) return; } } // Note: Letta does not emit `tool_return_message` over the wire in cloud // headless mode for auto-approved tools, and the relative ordering of // `auto_approval` vs the per-turn `stop_reason` is not stable. We therefore // defer the final "completed" flip to the terminating `result` message and // let `auto_approval` carry the tool from pending → in_progress in between. const update = wireToSessionUpdate(msg, state.id); if (!update) return; const u: any = update.update;
if (u?.sessionUpdate === "tool_call" && typeof u.toolCallId === "string") { const id = u.toolCallId; // Stash any diff content block so we can re-attach it when the tool // completes (the completion update carries the tool return text, which // would otherwise clobber the diff). if (Array.isArray(u.content)) { const diff = u.content.find((c: any) => c?.type === "diff"); if (diff) state.memoryDiffs.set(id, diff); } // Dedupe: Letta streams the same id across many arg deltas. Only the // first delta carries a real tool name; subsequent ones have name=null // and would clobber the title in the host UI. Convert later sightings // to tool_call_update so rawInput keeps refining without resetting the // row. if (state.seenToolCalls.has(id)) { // Coalesce: Letta streams 15–25 deltas per tool while args are // assembled. Drop the intermediate ones and only emit a refinement // when the rawInput is a complete JSON object (heuristic: not a // string, since complete args parse from the JSON string in // mapping.ts's safeParseJson). This cuts viewport churn dramatically. if (u.rawInput && typeof u.rawInput === "object") { // Memory + file-edit tools: the title-only initial tool_call had // args=null, so the diff couldn't be computed. Re-run decoration // here once args are complete and stash the diff for the eventual // completion. Both decorators need the RAW tool name (e.g. "Read", // "memory_replace") - titlesById may have been overwritten with a // descriptive title ("Read ~/foo.ts") by an earlier refinement, so // we prefer toolNamesById (set once on first sighting) for both. const rawName = state.toolNamesById.get(id) ?? u.title ?? ""; const enriched = decorateFileEditToolCall( decorateMemoryToolCall( { ...u, sessionUpdate: "tool_call_update" }, rawName, u.rawInput, ), rawName, u.rawInput, ); if (Array.isArray(enriched.content)) { const diff = enriched.content.find((c: any) => c?.type === "diff"); if (diff) state.memoryDiffs.set(id, diff); } const refined: any = { sessionId: state.id, update: { sessionUpdate: "tool_call_update", toolCallId: id, rawInput: enriched.rawInput ?? u.rawInput, ...(enriched.content ? { content: enriched.content } : {}), }, }; state.gate.emit(refined); } return; } state.seenToolCalls.add(id); state.inflightTools.add(id); if (typeof u.title === "string") { state.titlesById.set(id, u.title); // First tool_call carries the raw tool name as the title (descriptive // titles are applied later via tool_call_update); stash it for content // post-processing on completion. if (!state.toolNamesById.has(id)) state.toolNamesById.set(id, u.title); } if (shouldSuppressToolCallRow(String(u.title ?? ""))) { state.suppressedToolCalls.add(id); state.inflightTools.delete(id); return; } } if (u?.sessionUpdate === "tool_call_update") { const id = String(u.toolCallId); if (state.suppressedToolCalls.has(id)) { if (u.status === "completed" || u.status === "failed") { state.suppressedToolCalls.delete(id); } return; } const rawName = state.toolNamesById.get(id) ?? ""; // Strip ` 42→` line-number prefixes from Read tool returns so the // host renders file contents without the model-facing decoration. if (rawName === "Read" && Array.isArray(u.content)) { u.content = u.content.map((c: any) => { if (c?.type === "content" && c.content?.type === "text") { return { ...c, content: { ...c.content, text: stripReadLineNumbers(String(c.content.text ?? "")) }, }; } return c; }); } if ( u.status === "completed" && Array.isArray(u.content) && !shouldKeepSuccessfulToolReturnContent( rawName, u.content .map((c: any) => (c?.type === "content" ? String(c.content?.text ?? "") : "")) .join("\n"), ) ) { u.content = undefined; } if (u.status === "completed" || u.status === "failed") { state.inflightTools.delete(id); const isFileEdit = rawName === "Edit" || rawName === "Write" || rawName === "MultiEdit" || rawName === "ApplyPatch"; // Re-attach the stashed diff so the host UI shows what changed even // after the tool return body lands. For filesystem edits on success, // the tool return is just a noisy `{"message":"Successfully replaced // ..."}` JSON blob with no useful information - replace it with the // diff. For memory tools and for any failure, keep the return body // alongside the diff so the user can see status / error text. const diff = state.memoryDiffs.get(id); if (diff) { const existing = Array.isArray(u.content) ? u.content : []; const alreadyHasDiff = existing.some( (c: any) => c?.type === "diff" && c?.path === (diff as any).path, ); if (isFileEdit && u.status === "completed") { u.content = [diff]; } else if (!alreadyHasDiff) { u.content = [diff, ...existing]; } state.memoryDiffs.delete(id); } } } state.gate.emit(update); }
private async handleControlRequest(state: SessionState, msg: any): Promise<void> { const reqId = String(msg.request_id ?? ""); const inner = msg.request ?? {}; if (inner.subtype === "can_use_tool") { await this.handleCanUseTool(state, reqId, inner); return; } // Unknown subtype: reply with error so Letta doesn't hang. state.proc.sendRaw({ type: "control_response", response: { subtype: "error", request_id: reqId, error: `unsupported control_request subtype: ${inner.subtype}`, }, }); }
private async handleCanUseTool( state: SessionState, requestId: string, req: any, ): Promise<void> { const toolName = String(req.tool_name ?? "tool"); const toolCallId = String(req.tool_call_id ?? `call-${Date.now()}`); const input = req.input ?? {}; const diffs = Array.isArray(req.diffs) ? req.diffs : [];
// In-session auto-approval. If the user earlier picked "Always allow this // tool" and this call positively parses as benign, approve without // re-prompting. For Bash the grant is scoped to the leading program (e.g. // `linear`), never all-Bash, and the command must parse as safe (granted // program + read-only filter pipeline, no chaining/redirects/substitution); // any uncertainty falls through to a normal prompt. See permission-policy.ts. // Kill switch: LETTA_ACP_SESSION_GRANTS=0 disables the cache (no rebuild). const sessionGrantsEnabled = process.env.LETTA_ACP_SESSION_GRANTS !== "0"; if (sessionGrantsEnabled) { const auto = state.permissionGrants.autoApprove(toolName, input); if (auto.approved) { // Visible note so an auto-approval is never invisible - Matias reviews // his commands and wants to see when one ran without a prompt. state.gate.emit({ sessionId: state.id, update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: `\n_(auto-approved: ${auto.label})_\n` }, }, } as any); state.proc.sendRaw({ type: "control_response", response: { subtype: "success", request_id: requestId, response: { behavior: "allow" }, }, }); return; } }
// Build ACP request_permission. Surface diffs when present. Otherwise rely // on structured title + rawInput so the host does not fall back to a noisy // raw JSON text block. const content: ContentBlock[] = []; for (const d of diffs) { const diff = lettaDiffToAcp(d); if (diff) content.push({ type: "resource_link", uri: `diff:${diff.path}` } as any); } const acpReq: RequestPermissionRequest = buildPermissionRequestToolCall( state.id, toolCallId, toolName, input, content, );
let allow = false; let denyMessage = "User rejected the tool call."; let persistAlways = false; try { const resp: RequestPermissionResponse = await this.conn.requestPermission(acpReq); const outcome: any = (resp as any).outcome ?? {}; if (outcome.outcome === "selected") { const id = String(outcome.optionId ?? ""); allow = id === "allow-once" || id === "allow-always"; persistAlways = id === "allow-always"; // Remember an explicit "Always allow this tool" for the rest of the // session so matching calls stop re-prompting. For Bash this records // only the leading program token (e.g. `linear`), never all-Bash. if (persistAlways && sessionGrantsEnabled) { state.permissionGrants.recordAllowAlways(toolName, input); } } else if (outcome.outcome === "cancelled") { denyMessage = "User cancelled the permission request."; } } catch (err) { denyMessage = `permission request failed: ${err}`; }
// "Allow always" persistence: ask letta to write a permission rule so the // next session auto-allows without re-prompting. letta derives the rule // (with the same safety analysis as the interactive CLI) and owns the // settings-file write. Kill switch: set LETTA_ACP_PERSIST_PERMISSIONS=0 in // .env to disable without a rebuild. const persistEnabled = process.env.LETTA_ACP_PERSIST_PERMISSIONS !== "0"; const allowResponse: { behavior: "allow"; persist?: boolean } = { behavior: "allow", }; if (allow && persistAlways && persistEnabled) { allowResponse.persist = true; }
// Letta wire shape - see src/types/protocol.ts (ControlResponse + // ControlResponseBody). Envelope wraps a ControlResponseBody under // `response`; the success-variant body carries `request_id` and the // CanUseToolResponse payload as a nested `response`. Asymmetry vs // ControlRequest (which puts `request_id` on the envelope) is by design // upstream - don't "flatten" without re-reading the types. const body = allow ? { subtype: "success", request_id: requestId, response: allowResponse } : { subtype: "success", request_id: requestId, response: { behavior: "deny", message: denyMessage }, }; state.proc.sendRaw({ type: "control_response", response: body }); }
private completeInflightTools( state: SessionState, opts: { treatOrphansAs: "completed" | "failed"; note: string; }, ): void { if (state.inflightTools.size === 0) return; const ids = [...state.inflightTools]; state.inflightTools.clear(); for (const id of ids) { if (state.suppressedToolCalls.has(id)) { state.suppressedToolCalls.delete(id); continue; } state.gate.emit({ sessionId: state.id, update: { sessionUpdate: "tool_call_update", toolCallId: id, status: opts.treatOrphansAs, content: completionContentForOrphan({ status: opts.treatOrphansAs, note: opts.note, }), }, } as any); } }
/** Append a raw wire message to the trace file when LETTA_ACP_TRACE is set. */ private trace(msg: unknown): void { if (!TRACE_PATH) return; try { appendFileSync(TRACE_PATH, JSON.stringify({ ts: Date.now(), msg }) + "\n"); } catch { // Tracing is best-effort and must never affect the turn. } }
private clearTurnTimers(state: SessionState): void { if (state.quiescenceTimer) { clearTimeout(state.quiescenceTimer); state.quiescenceTimer = undefined; } if (state.maxHoldTimer) { clearTimeout(state.maxHoldTimer); state.maxHoldTimer = undefined; } }
/** Execute the timer/resolve actions produced by the quiescence tracker. */ private applyQuiescence(state: SessionState, actions: QuiescenceAction[]): void { for (const action of actions) { switch (action.type) { case "arm-quiescence-timer": { if (state.quiescenceTimer) clearTimeout(state.quiescenceTimer); if (QUIESCENCE_DEBOUNCE_MS <= 0) { // Debounce disabled: decide immediately. this.applyQuiescence(state, state.quiescence.onQuiescenceTimer()); } else { state.quiescenceTimer = setTimeout(() => { state.quiescenceTimer = undefined; this.applyQuiescence(state, state.quiescence.onQuiescenceTimer()); }, QUIESCENCE_DEBOUNCE_MS); } break; } case "cancel-quiescence-timer": if (state.quiescenceTimer) { clearTimeout(state.quiescenceTimer); state.quiescenceTimer = undefined; } break; case "arm-max-hold": if (state.maxHoldTimer) clearTimeout(state.maxHoldTimer); if (MAX_HOLD_MS > 0) { state.maxHoldTimer = setTimeout(() => { state.maxHoldTimer = undefined; process.stderr.write( `[letta-acp] turn held ${MAX_HOLD_MS}ms without queue drain; force-resolving\n`, ); this.applyQuiescence(state, state.quiescence.onMaxHold()); }, MAX_HOLD_MS); } break; case "cancel-max-hold": if (state.maxHoldTimer) { clearTimeout(state.maxHoldTimer); state.maxHoldTimer = undefined; } break; case "resolve": void this.finalizeTurn(state, action.reason).catch((err) => { process.stderr.write(`[letta-acp] finalizeTurn failed: ${err}\n`); }); break; } } }
/** * Close the ACP turn once the quiescence tracker decides it's done (queue * drained, max-hold cap, or cancel). Flushes pending updates first for the * same in-flight-write ordering guarantee resolvePrompt relies on. */ private async finalizeTurn(state: SessionState, reason: ResolveReason): Promise<void> { const pending = state.pending; if (!pending) return; state.pending = undefined; this.clearTurnTimers(state); await state.gate.flush(); const stopReason = reason === "cancelled" ? "cancelled" : "end_turn"; pending.resolve({ stopReason }); }
private async resolvePrompt(state: SessionState, msg: ResultWire): Promise<void> { const pending = state.pending; if (!pending) return; state.pending = undefined; if (msg.subtype === "error") { // Surface the run error text instead of a silent "refusal" with no // content. Letta carries the detail in `result`; without this the user // just sees a bare stop with no explanation (e.g. after a turn was // cancelled mid-tool-call and the next turn errors on stale state). const detail = (msg.result ?? "").trim() || "Letta run ended in error."; state.gate.emit({ sessionId: state.id, update: { sessionUpdate: "agent_message_chunk", content: { type: "text", text: `⚠️ ${detail}` }, }, } as any); } const stopReason = msg.subtype === "success" ? "end_turn" : msg.subtype === "interrupted" ? "cancelled" : "refusal"; // Drain every emitted `session/update` to the wire before sending the // prompt response. The SDK serializes all writes through one FIFO queue, so // once these settle the response is guaranteed to come last - no update can // arrive after the turn ends ("Out of turn - ACP server bug"). await state.gate.flush(); pending.resolve({ stopReason }); }
private async rejectPrompt(state: SessionState, msg: ErrorWire): Promise<void> { const pending = state.pending; if (!pending) { // Surface unsolicited errors via stderr. process.stderr.write(`[letta-acp] error: ${msg.message}\n`); return; } state.pending = undefined; // Flush any in-flight tool-completion updates (e.g. orphans marked failed) // before the prompt rejection closes the turn, for the same ordering // guarantee as resolvePrompt. await state.gate.flush(); pending.reject(new Error(msg.message)); }}
function extractText(params: PromptRequest): string { // Concatenate text blocks from ACP ContentBlock[]. const parts: string[] = []; for (const block of params.prompt ?? []) { if (block.type === "text") parts.push(block.text); else if (block.type === "resource_link" && (block as any).uri) parts.push((block as any).uri); } return parts.join("\n");}
function rpcError(code: number, message: string): Error & { code: number } { const e = new Error(message) as Error & { code: number }; e.code = code; return e;}
/** * Translate a Letta DiffPreview into a best-effort ACP Diff. * Letta's "advanced" mode carries hunks; "fallback" and "unpreviewable" only * carry a reason string. We render those as a single empty-hunk diff so the * host still sees the file path. */function lettaDiffToAcp(d: any): { path: string; oldText: string; newText: string } | null { if (!d || typeof d !== "object") return null; const path = String(d.fileName ?? ""); if (!path) return null; if (d.mode === "advanced" && Array.isArray(d.hunks)) { const oldLines: string[] = []; const newLines: string[] = []; for (const h of d.hunks) { for (const line of h.lines ?? []) { if (line.type === "context") { oldLines.push(line.content); newLines.push(line.content); } else if (line.type === "remove") { oldLines.push(line.content); } else if (line.type === "add") { newLines.push(line.content); } } } return { path, oldText: oldLines.join("\n"), newText: newLines.join("\n") }; } return { path, oldText: "", newText: `(${d.mode}: ${d.reason ?? "no preview"})` };}
function main(): void { const inputWeb = Readable.toWeb(process.stdin) as unknown as ReadableStream<Uint8Array>; const outputWeb = Writable.toWeb(process.stdout) as unknown as WritableStream<Uint8Array>; const stream = ndJsonStream(outputWeb, inputWeb); new AgentSideConnection((conn) => new LettaAcpAgent(conn), stream); process.stdin.resume();}
main();