diff --git a/cli/src/client.test.ts b/cli/src/client.test.ts index 2aeca359..1d53ac16 100644 --- a/cli/src/client.test.ts +++ b/cli/src/client.test.ts @@ -1,6 +1,6 @@ import assert from "node:assert/strict" import test from "node:test" -import { createChatClient, type StreamEvent } from "./client.js" +import { createChatClient, WorkspaceMissingError, type StreamEvent } from "./client.js" const page=(cursor:number,events:unknown[]) => `data: ${JSON.stringify({cursor,events})}\n\n` test("daemon thinking events stay a distinct type from assistant text",async()=>{ @@ -83,3 +83,79 @@ test("message timestamps survive the wire without inventing missing or invalid d await client.stream({ onEvent: event => events.push(event) }) assert.deepEqual(events, [user, { ...message, timestamp: user.timestamp + 1000 }, message, message]) }) + +test("unfinished argument previews are bounded across reconnects and recover from a snapshot", async () => { + for (const [count, text] of [[33, '{"code":"'], [3, 'x'.repeat(700_000)]] as const) { + const urls: string[] = [] + const client = createChatClient({ baseUrl: "http://localhost", agentId: "session", fetchImpl: async url => { + urls.push(String(url)) + const cursor = urls.length + return new Response(page(cursor, cursor <= count + ? [{ type: "arguments_delta", callId: `call-${cursor}`, text }] + : [{ type: "reset" }, { type: "message", role: "assistant", text: "recovered" }])) + } }) + for (let i = 1; i < count; i++) await client.stream({ onEvent: () => {} }) + await assert.rejects(client.stream({ onEvent: () => {} }), /previews exceed client limit/) + const events: StreamEvent[] = [] + await client.stream({ onEvent: event => events.push(event) }) + assert.match(urls.at(-1)!, /after_seq=-1$/) + assert.deepEqual(events.at(-1), { type: "message", role: "assistant", text: "recovered" }) + } +}) + +test("detaching drops argument previews and resets the reconnect cursor", async () => { + const controller = new AbortController() + const urls: string[] = [] + const client = createChatClient({ baseUrl: "http://localhost", agentId: "session", fetchImpl: async url => { + urls.push(String(url)) + return new Response(page(urls.length, [{ type: "arguments_delta", callId: "same", text: JSON.stringify({ code: urls.length === 1 ? "old()" : "new()" }) }])) + } }) + await client.stream({ signal: controller.signal, onEvent: event => { + if (event.type === "tool_progress" && event.progress) controller.abort() + } }) + const events: StreamEvent[] = [] + await client.stream({ onEvent: event => events.push(event) }) + assert.match(urls[1]!, /after_seq=-1$/) + assert(events.some(event => event.type === "tool_progress" && event.progress?.code?.text === "new()")) +}) + +test("canceling a chat lifetime aborts pending send and status requests", async () => { + const controller = new AbortController() + const client = createChatClient({ baseUrl: "http://localhost", agentId: "session", fetchImpl: async (_url, init) => { + const signal = init?.signal + assert(signal) + return new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => reject(signal.reason), { once: true }) + }) + } }) + const sent = client.send("hello", controller.signal) + const status = client.getStatus(controller.signal) + controller.abort() + await assert.rejects(sent, { name: "AbortError" }) + await assert.rejects(status, { name: "AbortError" }) +}) + + + + +test("send exposes a structured missing-workspace error", async () => { + const client = createChatClient({ baseUrl: "http://localhost", agentId: "session", fetchImpl: async () => + new Response(JSON.stringify({ code: "workspace_missing", error: "workspace not found", workspace: "/gone" }), { status: 409, headers: { "content-type": "application/json" } }), + }) + await assert.rejects(client.send("continue"), error => error instanceof WorkspaceMissingError && error.workspace === "/gone") +}) + +test("workspace replacement requires daemon capability and returns authoritative metadata", async () => { + const requests: { url: string; body?: string }[] = [] + const client = createChatClient({ baseUrl: "http://localhost", agentId: "session", fetchImpl: async (input, init) => { + requests.push({ url: String(input), ...(typeof init?.body === "string" ? { body: init.body } : {}) }) + if (String(input).endsWith("/health")) return Response.json({ capabilities: ["session_workspace"] }) + return Response.json({ workspace: "/actual/workspace", id: "session" }) + } }) + assert.deepEqual(await client.replaceWorkspace?.("/requested/workspace"), { workspace: "/actual/workspace" }) + assert.deepEqual(requests.map(request => request.url), ["http://localhost/health", "http://localhost/sessions/session/workspace"]) + assert.deepEqual(JSON.parse(requests[1]!.body!), { workspace: "/requested/workspace" }) + + const old = createChatClient({ baseUrl: "http://localhost", agentId: "session", fetchImpl: async () => Response.json({ capabilities: [] }) }) + await assert.rejects(old.replaceWorkspace!("/new"), /daemon upgrade needed/) +}) diff --git a/cli/src/client.ts b/cli/src/client.ts index 064e1871..75d611cd 100644 --- a/cli/src/client.ts +++ b/cli/src/client.ts @@ -20,13 +20,22 @@ export interface StreamOptions { export type AgentPhase = "resting" | "preparing" | "reasoning" | "tool" | "waiting" export type AgentStatus = { running: boolean; idle: boolean; phase?: AgentPhase } +export type WorkspaceUpdate = { workspace: string } + +export class WorkspaceMissingError extends Error { + readonly name = "WorkspaceMissingError" + constructor(readonly workspace: string) { + super(`workspace not found: ${workspace}`) + } +} export interface ChatClient { /** The sender identity used by send; lets views suppress their optimistic echo. */ readonly clientId?: string - send: (content: string) => Promise<{ ok: true }> + send: (content: string, signal?: AbortSignal) => Promise<{ ok: true }> + replaceWorkspace?: (workspace: string, signal?: AbortSignal) => Promise interrupt?: () => Promise<{ interrupted: boolean }> - getStatus: () => Promise + getStatus: (signal?: AbortSignal) => Promise stream: (options: StreamOptions) => Promise } @@ -154,15 +163,17 @@ const toMessage = (raw: unknown): StreamEvent | null => { return { type: "message", role: "assistant", text: record.content } } -const parseError = async (res: Response): Promise => { - const fallback = `${res.status} ${res.statusText}`.trim() +const responseError = async (res: Response): Promise => { + const fallback = `${res.status} ${res.statusText}`.trim() || "request failed" try { - const data = (await res.json()) as { error?: unknown } - if (typeof data.error === "string" && data.error.trim()) return data.error + const data = (await res.json()) as { code?: unknown; error?: unknown; workspace?: unknown } + if (data.code === "workspace_missing" && typeof data.workspace === "string") + return new WorkspaceMissingError(data.workspace) + if (typeof data.error === "string" && data.error.trim()) return new Error(data.error) } catch {} - return fallback || "request failed" + return new Error(fallback) } const splitLines = (chunkBuffer: string): { lines: string[]; rest: string } => { @@ -184,31 +195,53 @@ export function createChatClient(options: CreateChatClientOptions): ChatClient { const agentUrl = (path: string) => agentId ? url(`/sessions/${encodeURIComponent(agentId)}${path}`) : url(path) - const send: ChatClient["send"] = async (content) => { + const requestSignal = (signal?: AbortSignal): AbortSignal => signal + ? AbortSignal.any([signal, AbortSignal.timeout(20_000)]) : AbortSignal.timeout(20_000) + const send: ChatClient["send"] = async (content, signal) => { const res = await fetchImpl(agentUrl(agentId ? "/events" : "/trigger/chat"), { method: "POST", headers: headers(true), body: JSON.stringify({ content, ...(clientId ? { clientId } : {}) }), + signal: requestSignal(signal), }) if (!res.ok) { - throw new Error(await parseError(res)) + throw await responseError(res) } return { ok: true } } + const replaceWorkspace: ChatClient["replaceWorkspace"] = agentId ? async (workspace, signal) => { + const health = await fetchImpl(url("/health"), { headers: headers(), signal: requestSignal(signal) }) + if (!health.ok) throw await responseError(health) + const capabilities = (await health.json() as { capabilities?: unknown }).capabilities + if (!Array.isArray(capabilities) || !capabilities.includes("session_workspace")) + throw new Error("daemon upgrade needed to change this workspace; when ready, run albedo daemon --stop, then albedo (this clears python variables)") + + const res = await fetchImpl(agentUrl("/workspace"), { + method: "POST", + headers: headers(true), + body: JSON.stringify({ workspace }), + signal: requestSignal(signal), + }) + if (!res.ok) throw await responseError(res) + const data = await res.json() as { workspace?: unknown } + if (typeof data.workspace !== "string") throw new Error("daemon returned invalid session metadata") + return { workspace: data.workspace } + } : undefined + const interrupt = async (): Promise<{ interrupted: boolean }> => { const res = await fetchImpl(agentUrl(agentId ? "/interrupt" : "/awp/interrupt"), { method: "POST", headers: headers(), signal: AbortSignal.timeout(15_000) }) - if (!res.ok) throw new Error(await parseError(res)) + if (!res.ok) throw await responseError(res) const data = await res.json() as { interrupted?: unknown } return { interrupted: data.interrupted === true } } - const getStatus: ChatClient["getStatus"] = async () => { - const res = await fetchImpl(agentUrl("/status"), { headers: headers() }) + const getStatus: ChatClient["getStatus"] = async (signal) => { + const res = await fetchImpl(agentUrl("/status"), { headers: headers(), signal: requestSignal(signal) }) if (!res.ok) { - throw new Error(await parseError(res)) + throw await responseError(res) } const data = (await res.json()) as { running?: unknown; idle?: unknown; phase?: unknown } @@ -226,7 +259,7 @@ export function createChatClient(options: CreateChatClientOptions): ChatClient { : `/awp/stream?${afterSeq ? `after_seq=${afterSeq}` : "tail=true"}` const res = await fetchImpl(agentUrl(route), { signal, headers: headers() }) if (!res.ok || !res.body) { - throw new Error(res.ok ? "stream body missing" : await parseError(res)) + throw res.ok ? new Error("stream body missing") : await responseError(res) } onOpen?.() @@ -290,8 +323,14 @@ export function createChatClient(options: CreateChatClientOptions): ChatClient { } } if (item.type === "arguments_delta" && typeof item.callId === "string" && typeof item.text === "string") { + const fresh = !argumentsByCall.has(item.callId) + const retained = [...argumentsByCall].reduce((size, [id, args]) => size + id.length + args.length, 0) + if ((fresh && argumentsByCall.size >= 32) || retained + item.text.length + (fresh ? item.callId.length : 0) > 2_000_000) { + argumentsByCall.clear() + afterSeq = -1 + throw new Error("tool argument previews exceed client limit") + } const args = (argumentsByCall.get(item.callId) ?? "") + item.text - if (args.length > 2_000_000) throw new Error("tool arguments exceed client limit") argumentsByCall.set(item.callId, args) report({ id: item.callId, function: { name: "python", arguments: args } }, "generating") continue @@ -316,8 +355,9 @@ export function createChatClient(options: CreateChatClientOptions): ChatClient { } } } finally { - // Keep only unfinished raw arguments for cursor-based reconnection, never - // their decoded code or syntax tree while the transport is disconnected. + // Reconnects keep bounded unfinished arguments; detaching releases them + // and starts any future attachment from a fresh snapshot. + if (signal?.aborted) { argumentsByCall.clear(); afterSeq = -1 } signal?.removeEventListener("abort", cancel) await reader.cancel().catch(() => {}) reader.releaseLock() @@ -328,6 +368,7 @@ export function createChatClient(options: CreateChatClientOptions): ChatClient { return { clientId, send, + ...(replaceWorkspace ? { replaceWorkspace } : {}), interrupt, getStatus, stream, diff --git a/cli/src/ui/app.tsx b/cli/src/ui/app.tsx index 8901c640..0b013fb7 100644 --- a/cli/src/ui/app.tsx +++ b/cli/src/ui/app.tsx @@ -15,10 +15,20 @@ export function App({ connection, initial, workspace, quit, login = false }: { c const [choosing,setChoosing]=useState(!initial) const [choosingModel,setChoosingModel]=useState(false) const listedSessions = selected && !sessions.some(session => session.id === selected.id) ? [selected, ...sessions] : sessions - const changeModel = async (model: string): Promise => { + const workspaceChanged = (workspace: string): void => { + setSelected(current => current && { ...current, workspace }) + if (selected) setSessions(current => current.map(session => session.id === selected.id ? { ...session, workspace } : session)) + } + const changeModel = async (model: string, provider?: string): Promise => { if (!selected) return - await request(connection, `/sessions/${selected.id}/model`, { model }) - setSelected({ ...selected, model }); setChoosingModel(false); setError("") + if (provider && provider !== selected.provider) { + const health = await request<{ capabilities?: string[] }>(connection, "/health") + if (!health.capabilities?.includes("session_provider")) + throw new Error("daemon upgrade needed to switch providers; when ready, run albedo daemon --stop, then albedo (this clears python variables)") + } + const changed = await request>>(connection, `/sessions/${selected.id}/model`, { model, provider }) + setSelected({ ...selected, model: changed.model ?? model, provider: changed.provider ?? selected.provider, protocol: changed.protocol ?? selected.protocol }) + setChoosingModel(false); setError("") } useEffect(() => { if (!choosing) return @@ -31,7 +41,7 @@ export function App({ connection, initial, workspace, quit, login = false }: { c } if (loggingIn) return setLoggingIn(undefined)} onDone={name=>{ setLoggingIn(undefined); setError("") - setNotice(`${name} selected for new sessions${selected ? `; this session keeps ${selected.provider}` : ""}`) + setNotice(`${name} selected for new sessions${selected ? `; use /model to switch this session from ${selected.provider}` : ""}`) if (!selected && !sessions.length) create() }} /> return @@ -46,8 +56,8 @@ export function App({ connection, initial, workspace, quit, login = false }: { c ...listedSessions.map(session => ({ id:session.id,label:typeof session.title === "string" ? session.title.trim() || "new session" : "session · label unavailable",detail:`${session.model} · ${session.workspace}` })), ]} onSelect={id => { if (id==="new") create(); else if (id==="login") setLoggingIn({}); else { setSelected(listedSessions.find(session=>session.id===id));setChoosing(false) } }} onCancel={() => selected ? setChoosing(false) : quit()} /> : selected && <>setChoosing(true)} onQuit={quit} onCreate={create} - commands={[{ name:"/login",description:"add or select a named openai-compatible api" },{ name:"/new",description:"new coding session" },{ name:"/sessions",description:"switch session" },{ name:"/model",description:"choose this session's model" }]} + workspace={selected.workspace} model={selected.model} onWorkspaceChanged={workspaceChanged} onBack={()=>setChoosing(true)} onQuit={quit} onCreate={create} + commands={[{ name:"/login",description:"add or select a named openai-compatible api" },{ name:"/new",description:"new coding session" },{ name:"/sessions",description:"switch session" },{ name:"/model",description:"choose this session's provider and model" }]} onCommand={(value,clear)=> { if (value === "/login" || value.startsWith("/login ")) { clear(); setLoggingIn({ name: value.slice(6).trim() || undefined }); return true } if (value.startsWith("/model ")) { clear();setError("");void changeModel(value.slice(7).trim()).catch(error=>setError(String(error)));return true } diff --git a/cli/src/ui/chat.test.ts b/cli/src/ui/chat.test.ts index 53fb230f..0fbcb92a 100644 --- a/cli/src/ui/chat.test.ts +++ b/cli/src/ui/chat.test.ts @@ -4,7 +4,7 @@ import { PassThrough } from "node:stream" import { createElement, type ReactElement } from "react" import { render } from "ink" import { ChatScreen } from "./chat.js" -import type { StreamEvent, StreamOptions } from "../client.js" +import { WorkspaceMissingError, type StreamEvent, type StreamOptions } from "../client.js" const strip = (text: string): string => text.replace(/\x1b\[[0-9;?]*[A-Za-z]/g, "") const tick = (): Promise => new Promise((resolve) => { setTimeout(resolve, 5) }) const until = async (ready: () => boolean, what: string): Promise => { @@ -12,7 +12,7 @@ const until = async (ready: () => boolean, what: string): Promise => { throw new Error(`timed out waiting for ${what}`) } -type Screen = { last: () => string; all: () => string; raw: () => string; shows: (text: string) => Promise; write: (text: string) => void } +type Screen = { last: () => string; all: () => string; raw: () => string; shows: (text: string) => Promise; write: (text: string) => void; flush: () => Promise; unmount: () => void } /** Mount a screen on injected streams that look like a terminal to Ink. */ const mount = (node: ReactElement, t: { after: (fn: () => void) => void }, columns = 80): Screen => { @@ -30,6 +30,10 @@ const mount = (node: ReactElement, t: { after: (fn: () => void) => void }, colum shows: (text) => until(() => screen.all().includes(text), `screen text ${JSON.stringify(text)}`), raw: () => frames.join(""), write: (text) => { stdin.write(text) }, + // Two writes in one tick reach Ink as one paste-sized chunk; a person types + // the text and presses enter in separate turns, so flush between them. + flush: () => app.waitUntilRenderFlush(), + unmount: () => app.unmount(), } return screen } @@ -184,3 +188,143 @@ test("confirmed user and streamed assistant headings get clocks without duplicat assert.equal(screen.last().split("question").length - 1, 1) assert.equal(screen.last().split("the answer").length - 1, 1) }) + +test("missing workspace asks for a replacement and retries only after confirmation", async t => { + const sent: string[] = [] + const replacements: string[] = [] + const updated: string[] = [] + const screen = mount(createElement(ChatScreen, { + transport: { + send: async content => { + sent.push(content) + if (sent.length === 1) throw new WorkspaceMissingError("/old/missing") + return { ok: true as const } + }, + replaceWorkspace: async workspace => { replacements.push(workspace); return { workspace: `${workspace}/` } }, + getStatus: async () => ({ running: false, idle: true }), + stream: async options => { options.onOpen?.(); await new Promise(resolve => options.signal?.addEventListener("abort", () => resolve(), { once: true })) }, + }, + workspace: "/old/missing", agentName: "albedo", onWorkspaceChanged: workspace => { updated.push(workspace) }, + onBack: () => {}, onQuit: () => {}, settleMs: 20, + }), t) + + await until(() => screen.last().includes("ready"), "chat ready") + screen.write("finish the task") + await screen.flush() + screen.write("\r") + await until(() => screen.last().includes("workspace not found: /old/missing"), "workspace recovery form") + assert.deepEqual(sent, ["finish the task"]) + assert.deepEqual(replacements, []) + screen.write("\u0015") + await screen.flush() + screen.write("/new/workspace") + await screen.flush() + screen.write("\r") + await until(() => sent.length === 2, "original prompt retried") + assert.deepEqual(replacements, ["/new/workspace"]) + assert.deepEqual(updated, ["/new/workspace/"], "session metadata comes from the daemon") + assert.deepEqual(sent, ["finish the task", "finish the task"]) + await until(() => !screen.last().includes("workspace not found"), "recovery form closed") +}) + +test("canceling workspace recovery preserves the chat and unsent prompt", async t => { + let stream: StreamOptions | undefined + let replacements = 0 + const screen = mount(createElement(ChatScreen, { + transport: { + send: async () => { throw new WorkspaceMissingError("/gone") }, + replaceWorkspace: async workspace => { replacements++; return { workspace } }, + getStatus: async () => ({ running: false, idle: true }), + stream: async options => { stream = options; options.onOpen?.(); await new Promise(resolve => options.signal?.addEventListener("abort", () => resolve(), { once: true })) }, + }, + workspace: "/gone", agentName: "albedo", onBack: () => {}, onQuit: () => {}, settleMs: 20, + }), t) + await until(() => stream !== undefined, "stream connected") + stream!.onEvent({ type: "message", role: "assistant", text: "existing chat stays here" }) + await screen.shows("existing chat stays here") + await until(() => screen.last().includes("ready"), "chat ready") + screen.write("unsent work") + await screen.flush() + screen.write("\r") + await until(() => screen.last().includes("workspace not found: /gone"), "workspace recovery form") + screen.write("\u001b") + await until(() => !screen.last().includes("workspace not found"), "recovery canceled") + assert(screen.last().includes("existing chat stays here")) + assert(screen.last().includes("unsent work")) + assert.equal(replacements, 0) +}) + +test("a rejected workspace stays open with the daemon's reason and the typed path", async t => { + const attempts: string[] = [] + const screen = mount(createElement(ChatScreen, { + transport: { + send: async () => { throw new WorkspaceMissingError("/gone") }, + replaceWorkspace: async workspace => { + attempts.push(workspace) + throw new Error("workspace must be an existing absolute directory") + }, + getStatus: async () => ({ running: false, idle: true }), + stream: async options => { options.onOpen?.(); await new Promise(resolve => options.signal?.addEventListener("abort", () => resolve(), { once: true })) }, + }, + workspace: "/gone", agentName: "albedo", onBack: () => {}, onQuit: () => {}, settleMs: 20, + }), t) + await until(() => screen.last().includes("ready"), "chat ready") + screen.write("retry me") + await screen.flush() + screen.write("\r") + await until(() => screen.last().includes("workspace not found: /gone"), "workspace recovery form") + screen.write("\u0015") + await screen.flush() + screen.write("/relative") + await screen.flush() + screen.write("\r") + await until(() => screen.last().includes("workspace must be an existing absolute directory"), "daemon reason") + assert(screen.last().includes("new workspace › /relative")) + screen.write("\u0015") + await screen.flush() + screen.write("/absolute/path") + await screen.flush() + screen.write("\r") + await until(() => attempts.length === 2, "corrected path retried") + assert.deepEqual(attempts, ["/relative", "/absolute/path"]) +}) + +test("mouse reports never edit the workspace path while recovery is open", async t => { + const screen = mount(createElement(ChatScreen, { + transport: { + send: async () => { throw new WorkspaceMissingError("/gone") }, + replaceWorkspace: async workspace => ({ workspace }), + getStatus: async () => ({ running: false, idle: true }), + stream: async options => { options.onOpen?.(); await new Promise(resolve => options.signal?.addEventListener("abort", () => resolve(), { once: true })) }, + }, + workspace: "/gone", agentName: "albedo", onBack: () => {}, onQuit: () => {}, settleMs: 20, + }), t) + await until(() => screen.last().includes("ready"), "chat ready") + screen.write("retry me") + await screen.flush() + screen.write("\r") + await until(() => screen.last().includes("workspace not found: /gone"), "workspace recovery form") + screen.write("\u001b[<64;10;10M") + await screen.flush() + screen.write("x") + await until(() => screen.last().includes("new workspace › /gonex"), "path holds only typed text") +}) + +test("unmount aborts a hung status request", async t => { + let requested = false + let aborted = false + const screen = mount(createElement(ChatScreen, { + transport: { + send: async () => ({ ok: true as const }), + getStatus: signal => new Promise((_resolve, reject) => { + requested = true + signal?.addEventListener("abort", () => { aborted = true; reject(signal.reason) }, { once: true }) + }), + stream: async options => { options.onOpen?.(); await new Promise(resolve => options.signal?.addEventListener("abort", () => resolve(), { once: true })) }, + }, + agentName: "albedo", onBack: () => {}, onQuit: () => {}, settleMs: 20, + }), t) + await until(() => requested, "status request") + screen.unmount() + await until(() => aborted, "status abort") +}) diff --git a/cli/src/ui/chat.tsx b/cli/src/ui/chat.tsx index a4fad2ef..2980805f 100644 --- a/cli/src/ui/chat.tsx +++ b/cli/src/ui/chat.tsx @@ -2,7 +2,7 @@ import { useEffect, useMemo, useRef, useState } from "react" import { setTimeout as delay } from "node:timers/promises" import { Box, Text, useApp, useInput, useWindowSize, useStdout, useAnimation, useIsScreenReaderEnabled } from "ink" import { homedir } from "node:os" -import { createChatClient, type AgentStatus, type ChatClient, type StreamEvent } from "../client.js" +import { createChatClient, WorkspaceMissingError, type AgentStatus, type ChatClient, type StreamEvent } from "../client.js" import { advanceCodeLine, type CodeLine } from "./code-line.js" import { MouseInput, type MouseEvent } from "./mouse.js" import { copyText } from "./clipboard.js" @@ -28,6 +28,7 @@ export type ChatScreenProps = { workspace?: string model?: string notice?: string + onWorkspaceChanged?: (workspace: string) => void copySelection?: (text: string) => Promise fetchImpl?: typeof fetch onBack: () => void @@ -39,6 +40,7 @@ export type ChatScreenProps = { type Drag = { rows: Rows; selection: Selection; top: number; column: number; direction: number } type Active = { kind: "text" | "thinking"; content: MarkdownIndex } +type WorkspaceRecovery = { missing: string; replacement: string; prompt: string; saving: boolean; error?: string } const SPINNER = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"] type Connection = "connecting" | "live" | "disconnected" @@ -73,7 +75,7 @@ const describeFailure = (error: unknown, agentId?: string): string => { export function ChatScreen({ baseUrl, transport, visible = true, commands, onCommand, onCreate, - token, agentId, agentName, workspace, model, notice, fetchImpl, copySelection, onBack, onQuit, + token, agentId, agentName, workspace, model, notice, onWorkspaceChanged, fetchImpl, copySelection, onBack, onQuit, settleMs = 750, reconnectMs = 1_000, }: ChatScreenProps) { const speaker = agentName?.trim() || agentId?.trim() || "agent" @@ -84,6 +86,7 @@ export function ChatScreen({ const [codeLine, setCodeLine] = useState(null) const eventRevision = useRef(0) const [draft, setDraft] = useState("") + const [workspaceRecovery, setWorkspaceRecovery] = useState() const [status, setStatus] = useState() const [failure, setFailure] = useState("") const [stopping, setStopping] = useState(false) @@ -108,6 +111,7 @@ export function ChatScreen({ const streamedEntries = useRef([]) const pendingUsers = useRef<{ index: number; text: string }[]>([]) const live = useRef(true) + const requests = useRef(undefined) const clientId = useRef(`cli-${crypto.randomUUID()}`) const client = useMemo(() => { if (transport || !baseUrl) return transport @@ -249,6 +253,7 @@ export function ChatScreen({ live.current = true if (!client) return const controller = new AbortController() + requests.current = controller let reported = "" const disconnected = (error: unknown): void => { if (controller.signal.aborted) return @@ -276,7 +281,7 @@ export function ChatScreen({ while (!controller.signal.aborted) { try { const revision = eventRevision.current - const next = await client.getStatus() + const next = await client.getStatus(controller.signal) if (controller.signal.aborted) return // A response requested before a stream update cannot erase newer activity. if (revision !== eventRevision.current) { @@ -300,6 +305,35 @@ export function ChatScreen({ return () => { live.current = false; controller.abort() } }, [client, agentId, settleMs, reconnectMs]) + const sendTurn = (value: string): void => { + const trimmed = value.trim() + if (!client || !trimmed) return + setDraft("") + settle() + setScrollTop(null) + setFailure("") + setStopped(false) + setUsage(undefined) + const index = push({ kind: "user", source: "you", text: trimmed }) + if (index !== undefined) pendingUsers.current.push({ index, text: trimmed }) + setStatus({ running: true, idle: false, phase: "preparing" }) + const sent = client.send(trimmed, requests.current?.signal) + sending.current = sent + void sent.catch((error) => { + if (!live.current) return + pendingUsers.current = pendingUsers.current.filter(item => item.index !== index) + if (error instanceof WorkspaceMissingError) { + if (index !== undefined && transcript.discard(index)) setRevision(current => current + 1) + setStatus({ running: false, idle: true, phase: "resting" }) + setDraft(current => current || value) + setWorkspaceRecovery({ missing: error.workspace, replacement: error.workspace, prompt: value, saving: false }) + return + } + failTurn(error) + if (live.current) setDraft((current) => current || value) + }).finally(() => { if (sending.current === sent) sending.current = null }) + } + const submit = (value: string): void => { const trimmed = value.trim() if (onCommand?.(trimmed, () => setDraft(""))) return @@ -308,7 +342,7 @@ export function ChatScreen({ if (trimmed === "/q" || trimmed === "/quit" || trimmed === "/exit") return onQuit() if (trimmed === "/status") { setDraft("") - void client?.getStatus().then((status) => push({ kind: "note", text: `${awakeness(status, speaker)}${workspace ? `\nworkspace: ${workspace}` : ""}${model ? `\nmodel: ${model}` : ""}${usage ? `\n${formatUsage(usage)}` : ""}` })).catch(failTurn) + void client?.getStatus(requests.current?.signal).then((status) => push({ kind: "note", text: `${awakeness(status, speaker)}${workspace ? `\nworkspace: ${workspace}` : ""}${model ? `\nmodel: ${model}` : ""}${usage ? `\n${formatUsage(usage)}` : ""}` })).catch(failTurn) return } if (["/v", "/verbose", "/t", "/thinking"].includes(trimmed)) { @@ -322,23 +356,26 @@ export function ChatScreen({ setCopyStatus("wheel scrolling and drag-to-copy are always on") return } - if (!client) return - setDraft("") - settle() - setScrollTop(null) - setFailure("") - setStopped(false) - setUsage(undefined) - const index = push({ kind: "user", source: "you", text: trimmed }) - if (index !== undefined) pendingUsers.current.push({ index, text: trimmed }) - setStatus({ running: true, idle: false, phase: "preparing" }) - const sent = client.send(trimmed) - sending.current = sent - void sent.catch((error) => { - pendingUsers.current = pendingUsers.current.filter(item => item.index !== index) - failTurn(error) - if (live.current) setDraft((current) => current || value) - }).finally(() => { if (sending.current === sent) sending.current = null }) + sendTurn(value) + } + + const replaceWorkspace = (replacement: string): void => { + const next = replacement.trim() + const recovery = workspaceRecovery + if (!recovery || recovery.saving || !next) return + if (!client?.replaceWorkspace) { + setWorkspaceRecovery({ ...recovery, error: "this connection cannot change a session workspace" }) + return + } + setWorkspaceRecovery({ ...recovery, replacement, saving: true, error: undefined }) + void client.replaceWorkspace(next, requests.current?.signal).then(updated => { + if (!live.current) return + onWorkspaceChanged?.(updated.workspace) + setWorkspaceRecovery(undefined) + sendTurn(recovery.prompt) + }).catch(error => { + if (live.current) setWorkspaceRecovery({ ...recovery, replacement, saving: false, error: message(error) }) + }) } const stop = (): void => { @@ -381,9 +418,10 @@ export function ChatScreen({ const nextCodeLine = advanceCodeLine(codeLine, toolProgress, screenReader ? 0 : progressCodeWidth(toolProgress, contentWidth)) // Adjust on a new sample/width before painting; animation ticks don't consume code again. if (nextCodeLine !== codeLine) setCodeLine(nextCodeLine) - // Chrome owns six rows. Menus borrow transcript space, never composer space. - const menuRows = Math.min(menu.rows, Math.max(0, height - 9)) - const transcriptRows = Math.max(1, height - 6 - (notice ? 1 : 0) - menuRows) + // Chrome owns six rows. Recovery replaces the composer and borrows two more. + const recoveryRows = workspaceRecovery ? 2 : 0 + const menuRows = Math.min(menu.rows, Math.max(0, height - 9 - recoveryRows)) + const transcriptRows = Math.max(1, height - 6 - recoveryRows - (notice ? 1 : 0) - menuRows) const history = transcript.layout(flags, speaker, contentWidth, viewportWidth) // Resizing or expanding changes row counts above the viewport, not the record // being read. Tail appends keep the same layout and never move a history anchor. @@ -538,23 +576,43 @@ export function ChatScreen({ {!inHistory && !failure && width >= 60 && {stats}} {"─".repeat(Math.max(1, width - padding * 2))} - - › - { - const events = mouseInput.read(input) - if (events !== null) { for (const event of events) handleMouse(event); return true } - if (key.escape && drag.current) { drag.current = null; setSelecting(false); redrawSelection(); return true } - if (menu.onKey(input, key, replace)) return true - if (key.upArrow || key.downArrow) { scroll(key.upArrow ? -1 : 1); return true } - if (key.pageUp || key.pageDown) { scroll(key.pageUp ? -transcriptRows : transcriptRows); return true } - if (key.ctrl && key.home) { setScrollTop(0); paintInput(); return true } - if (key.ctrl && key.end) { setScrollTop(null); paintInput(); return true } - if (key.escape) { stop(); return true } - return false - }} width={Math.max(1, width - padding * 2 - 2)} /> - - {menuRows > 0 && {menu.view}} + {workspaceRecovery ? <> + workspace not found: {workspaceRecovery.missing} + + new workspace › + setWorkspaceRecovery(current => current && { ...current, replacement, error: undefined })} + onSubmit={replaceWorkspace} onKey={(input, key) => { + // The composer is unmounted here but mouse reporting is not, and a + // wheel or drag report would otherwise land in the path. + if (mouseInput.read(input) !== null) return true + if (!key.escape) return false + setWorkspaceRecovery(undefined) + return true + }} width={Math.max(1, width - padding * 2 - 16)} /> + + + {workspaceRecovery.error ?? (workspaceRecovery.saving ? "updating workspace…" : "enter confirms and retries · esc cancels")} + + : <> + + › + { + const events = mouseInput.read(input) + if (events !== null) { for (const event of events) handleMouse(event); return true } + if (key.escape && drag.current) { drag.current = null; setSelecting(false); redrawSelection(); return true } + if (menu.onKey(input, key, replace)) return true + if (key.upArrow || key.downArrow) { scroll(key.upArrow ? -1 : 1); return true } + if (key.pageUp || key.pageDown) { scroll(key.pageUp ? -transcriptRows : transcriptRows); return true } + if (key.ctrl && key.home) { setScrollTop(0); paintInput(); return true } + if (key.ctrl && key.end) { setScrollTop(null); paintInput(); return true } + if (key.escape) { stop(); return true } + return false + }} width={Math.max(1, width - padding * 2 - 2)} /> + + {menuRows > 0 && {menu.view}} + } } diff --git a/cli/src/ui/login.test.ts b/cli/src/ui/login.test.ts index d241a52b..d5fdc651 100644 --- a/cli/src/ui/login.test.ts +++ b/cli/src/ui/login.test.ts @@ -37,7 +37,11 @@ test("first login masks the key, discovers models and saves a named provider", a let chosen: string | undefined const app = render(createElement(Login, { onDone: name => { chosen = name }, onCancel: () => assert.fail("unexpected cancel") }), { stdin, stdout, patchConsole: false, exitOnCtrlC: false }) t.after(() => app.unmount()) - const press = async (keys: string): Promise => { (stdin as unknown as PassThrough).write(keys); await new Promise(resolve => setTimeout(resolve, 20)) } + const press = async (keys: string): Promise => { + await app.waitUntilRenderFlush() + ;(stdin as unknown as PassThrough).write(keys) + await app.waitUntilRenderFlush() + } await until(() => painted.includes("provider name:")) await press("fixture"); await press("\r") await until(() => painted.includes("api base url:")) diff --git a/cli/src/ui/model-picker.test.tsx b/cli/src/ui/model-picker.test.tsx index cf150ac6..f3345813 100644 --- a/cli/src/ui/model-picker.test.tsx +++ b/cli/src/ui/model-picker.test.tsx @@ -15,7 +15,7 @@ const until = async (ready: () => boolean, what: string): Promise => { throw new Error(`timed out: ${what}`) } -test("/model opens the session provider's picker, saves selections and offers manual fallback without losing chat", { timeout: 15_000 }, async t => { +test("/model switches saved providers and models with manual fallback without losing chat", { timeout: 15_000 }, async t => { const home = await mkdtemp(join(tmpdir(), "albedo-model-")) const previousHome = process.env.ALBEDO_HOME process.env.ALBEDO_HOME = home @@ -24,22 +24,28 @@ test("/model opens the session provider's picker, saves selections and offers ma else process.env.ALBEDO_HOME = previousHome await rm(home, { recursive: true, force: true }) }) - let listed = 0, streams = 0, unavailable = false, rejectChange = false - const changes: string[] = [] + let listed = 0, streams = 0, unavailable = false, rejectChange = false, supportsProviders = false + const changes: { model: string; provider?: string }[] = [] const session = { id: "session", title: "existing conversation", workspace: "/tmp", provider: "saved", protocol: "responses", model: "old-model" } const server = createServer(async (req, res) => { - if (req.url === "/saved/models") { + if (req.url === "/health") { + res.end(JSON.stringify({ ok: true, version: 2, ...(supportsProviders ? { capabilities: ["session_provider"] } : {}) })) + } else if (req.url === "/saved/models") { listed++ assert.equal(req.headers.authorization, "Bearer saved-key") res.writeHead(unavailable ? 503 : 200, { "content-type": "application/json" }) res.end(JSON.stringify({ data: [{ id: "old-model" }, { id: "new-model" }] })) + } else if (req.url === "/other/models") { + assert.equal(req.headers.authorization, "Bearer other-key") + res.end(JSON.stringify({ data: [{ id: "other-model" }] })) } else if (req.url === "/sessions/session/model") { let body = "" for await (const chunk of req) body += chunk - const model = (JSON.parse(body) as { model: string }).model + const change = JSON.parse(body) as { model: string; provider?: string } if (rejectChange) { res.writeHead(409); res.end(JSON.stringify({ error: "session must be idle" })); return } - changes.push(model); session.model = model - res.end(JSON.stringify({ ok: true })) + changes.push(change); session.model = change.model + if (change.provider) { session.provider = change.provider; session.protocol = change.provider === "other" ? "chat_completions" : "responses" } + res.end(JSON.stringify({ ok: true, model: session.model, provider: session.provider, protocol: session.protocol })) } else if (req.url?.startsWith("/sessions/session/stream")) { streams++ res.writeHead(200, { "content-type": "text/event-stream" }) @@ -53,7 +59,7 @@ test("/model opens the session provider's picker, saves selections and offers ma assert(address && typeof address !== "string") const config = JSON.stringify({ active: "other", providers: { saved: { baseUrl: `http://127.0.0.1:${address.port}/saved`, apiKey: "saved-key", model: "old-model", protocol: "responses" }, - other: { baseUrl: `http://127.0.0.1:${address.port}/wrong`, apiKey: "other-key", model: "other-model", protocol: "responses" }, + other: { baseUrl: `http://127.0.0.1:${address.port}/other`, apiKey: "other-key", model: "other-model", protocol: "chat_completions" }, } }) await writeFile(join(home, "config.json"), config) // profiles captures ALBEDO_HOME on import; this test file has its own process. @@ -68,7 +74,7 @@ test("/model opens the session provider's picker, saves selections and offers ma const open = async (): Promise => { painted = "" await enter("/model") - await until(() => painted.includes("session model"), "model picker") + await until(() => painted.includes("session model") && painted.includes("current") && painted.includes(unavailable ? "could not list models" : "new-model"), "model picker") assert(painted.includes("current")) } await until(() => painted.includes("keep this conversation visible"), "chat") @@ -76,7 +82,7 @@ test("/model opens the session provider's picker, saves selections and offers ma painted = "" await enter("new-model") await until(() => changes.length === 1 && painted.includes("keep this conversation visible"), "selection returns to preserved chat") - assert.deepEqual(changes, ["new-model"]) + assert.deepEqual(changes, [{ model: "new-model", provider: "saved" }]) assert(painted.includes("new-model")) await open() painted = "" @@ -98,8 +104,41 @@ test("/model opens the session provider's picker, saves selections and offers ma painted = "" stdin.write("\r") await until(() => changes.length === 2 && painted.includes("keep this conversation visible"), "retry manual selection") - assert.deepEqual(changes, ["new-model", "manual-model"]) + assert.deepEqual(changes, [{ model: "new-model", provider: "saved" }, { model: "manual-model", provider: "saved" }]) assert.equal(listed, 3) - assert.equal(streams, 1, "switching models must not discard the mounted conversation") + await open() + await enter("change provider") + await until(() => painted.includes("session provider") && painted.includes("chat_completions"), "provider picker") + await enter("other") + await until(() => painted.includes("other-model"), "other provider models") + painted = "" + stdin.write("\x1b") + await until(() => painted.includes("keep this conversation visible"), "cancel provider change") + assert.equal(changes.length, 2, "browsing providers must not save a change") + await open() + await enter("change provider") + await until(() => painted.includes("chat_completions"), "provider picker again") + await enter("other") + await until(() => painted.includes("other-model"), "other provider models again") + painted = "" + stdin.write("\r") + await until(() => painted.includes("daemon upgrade needed"), "old daemon requires upgrade before changing provider") + assert.equal(changes.length, 2, "old daemon must not receive a model from another provider") + supportsProviders = true + rejectChange = true + stdin.write("\r") + await until(() => painted.includes("session must be idle"), "busy session rejects provider switch") + assert.equal(changes.length, 2) + rejectChange = false + painted = "" + stdin.write("\r") + await until(() => changes.length === 3 && painted.includes("keep this conversation visible"), "provider selection preserves chat") + assert.deepEqual(changes[2], { model: "other-model", provider: "other" }) + painted = "" + await enter("/model") + await until(() => painted.includes("albedo /model · other") && painted.includes("current"), "reopen on selected provider") + assert(!painted.includes("manual-model"), "old provider's model must not leak into the new catalog") + stdin.write("\x1b") + assert.equal(streams, 1, "switching providers must not discard the mounted conversation") assert.equal(await readFile(join(home, "config.json"), "utf8"), config, "session selection must not change provider defaults") }) diff --git a/cli/src/ui/model-picker.tsx b/cli/src/ui/model-picker.tsx index 5d8542d2..4df616b6 100644 --- a/cli/src/ui/model-picker.tsx +++ b/cli/src/ui/model-picker.tsx @@ -1,38 +1,70 @@ import { useEffect, useState } from "react" import { Box, Text, useWindowSize } from "ink" -import { modelNames, profiles } from "../profiles.js" +import { modelNames, profiles, type Profiles, type Settings } from "../profiles.js" import { Picker } from "./picker.js" import { TextInput } from "./text-input.js" -export function ModelPicker({ provider, current, onSelect, onCancel }: { - provider: string; current: string; onSelect: (model: string) => Promise; onCancel: () => void +type ModelPickerProps = { + provider: string; current: string; onSelect: (model: string, provider: string) => Promise; onCancel: () => void +} + +export function ModelPicker({ provider, current, onSelect, onCancel }: ModelPickerProps) { + const [saved, setSaved] = useState() + const [name, setName] = useState(provider) + const [choosingProvider, setChoosingProvider] = useState(false) + const [error, setError] = useState("") + const sessionProvider = provider || saved?.active + useEffect(() => { + let active = true + void profiles().then(saved => { + if (active) { setSaved(saved); setName(provider || saved.active || "") } + }).catch(() => { if (active) setError("could not read providers; check /login") }) + return () => { active = false } + }, [provider]) + return + albedo /model · {name || "session provider"} + changes this idle session only · provider defaults stay unchanged + {!saved ? <> + {error || "loading providers…"} + {}} onKey={(input, key) => { + if (key.escape || (key.ctrl && ["c", "d"].includes(input))) { onCancel(); return true } + return false + }} /> + : choosingProvider ? ({ id, label: id, detail: `${settings.protocol}${id === sessionProvider ? " · current" : ""}` })) + } onSelect={name => { setName(name); setChoosingProvider(false) }} onCancel={() => setChoosingProvider(false)} /> : + setChoosingProvider(true)} />} + +} + +function ProviderModels({ provider, settings, current, onSelect, onCancel, onProvider }: Omit & { + settings?: Settings; current?: string; onProvider: () => void }) { + const initial = current ?? settings?.model ?? "" const [catalog, setCatalog] = useState<{ names: string[]; note?: string }>() const [manual, setManual] = useState(false) - const [value, setValue] = useState(current) + const [value, setValue] = useState(initial) const [saving, setSaving] = useState(false) const [error, setError] = useState("") const { columns } = useWindowSize() useEffect(() => { const controller = new AbortController() - void (async () => { - const saved = await profiles() - const name = provider || saved.active - const settings = name && saved.providers[name] - if (!settings) throw new Error("session provider missing") - return modelNames(settings.baseUrl, settings.apiKey, controller.signal) - })().then(names => { + void (settings + ? modelNames(settings.baseUrl, settings.apiKey, AbortSignal.any([controller.signal, AbortSignal.timeout(10_000)])) + : Promise.reject(new Error("session provider missing")) + ).then(names => { if (!controller.signal.aborted) setCatalog({ names, ...(!names.length ? { note: "no models listed; enter a model id manually" } : {}) }) }).catch(() => { if (!controller.signal.aborted) setCatalog({ names: [], note: "could not list models for this provider; enter a model id manually or check /login" }) }) return () => controller.abort() - }, [provider]) + }, [settings]) const select = async (value: string): Promise => { const model = value.trim() if (!model || model.length > 512 || /[\x00-\x1f\x7f]/.test(model)) { setError("enter a model id of 1–512 characters"); return } setError(""); setSaving(true) - try { await onSelect(model) } + try { await onSelect(model, provider) } catch (error) { setError(error instanceof Error ? error.message : "could not change model"); setSaving(false) } } const cancelKey = (input: string, key: { escape: boolean; ctrl: boolean }): boolean => { @@ -40,21 +72,18 @@ export function ModelPicker({ provider, current, onSelect, onCancel }: { return false } return - albedo /model · {provider || "session provider"} - changes this idle session only · endpoint and protocol stay unchanged {error && {error}} - {saving ? switching model… : !catalog ? <> - loading models… - {}} onKey={cancelKey} /> - : manual ? <> + {saving ? switching model… : manual ? <> model id: { void select(value) }} onKey={cancelKey} width={Math.max(1, columns - 10)} /> enter choose · esc cancel : <> - {catalog.note && {catalog.note}} - ({ id: `model:${model}`, label: model, ...(model === current ? { detail: "current" } : {}) })), + {!catalog && loading models…} + {catalog?.note && {catalog.note}} + ({ id: `model:${model}`, label: model, ...(model === current ? { detail: "current" } : {}) })), { id: "manual", label: "enter model id manually" }, - ]} onSelect={id => { if (id === "manual") setManual(true); else void select(id.slice(6)) }} onCancel={onCancel} /> + ]} onSelect={id => { if (id === "provider") onProvider(); else if (id === "manual") setManual(true); else void select(id.slice(6)) }} onCancel={onCancel} /> } } diff --git a/cli/src/ui/transcript-index.ts b/cli/src/ui/transcript-index.ts index 4da56d3f..6ea67651 100644 --- a/cli/src/ui/transcript-index.ts +++ b/cli/src/ui/transcript-index.ts @@ -42,6 +42,14 @@ export class TranscriptIndex { return start } + /** Remove a provisional tail entry that the daemon did not accept. */ + discard(index: number): boolean { + if (index !== this.entries.length - 1) return false + this.entries.pop() + this.layouts.clear() + return true + } + timestamp(index: number, timestamp: number): void { const entry = this.entries[index] if (!entry || (entry.kind !== "user" && entry.kind !== "assistant")) return diff --git a/src/albedo/daemon/conversation.gleam b/src/albedo/daemon/conversation.gleam index c0966901..331f80bf 100644 --- a/src/albedo/daemon/conversation.gleam +++ b/src/albedo/daemon/conversation.gleam @@ -27,7 +27,7 @@ pub fn initialise(store: store.Store) -> Result(Nil, String) { store.query(store, fn(db) { use _ <- result.try( sqlight.exec( - "CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);", + "CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER,provider TEXT); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);", db, ) |> result.map_error(fn(e) { e.message }), @@ -103,6 +103,12 @@ pub fn initialise(store: store.Store) -> Result(Nil, String) { sqlight.exec("ALTER TABLE transcript ADD COLUMN timestamp INTEGER", db) |> result.map_error(fn(e) { e.message }) }) + use _ <- result.try(case list.contains(transcript_columns, "provider") { + True -> Ok(Nil) + False -> + sqlight.exec("ALTER TABLE transcript ADD COLUMN provider TEXT", db) + |> result.map_error(fn(e) { e.message }) + }) use _ <- result.try(recover_sessions(db)) sqlight.exec( "CREATE INDEX IF NOT EXISTS sessions_activity ON sessions(activity_seq DESC)", @@ -346,13 +352,14 @@ pub fn load_entries( store.query(store, fn(db) { use rows <- result.try( sqlight.query( - "SELECT payload,timestamp FROM transcript WHERE session=? ORDER BY seq", + "SELECT payload,timestamp,provider FROM transcript WHERE session=? ORDER BY seq", db, [sqlight.text(id)], { use payload <- decode.field(0, decode.bit_array) use timestamp <- decode.field(1, decode.optional(decode.int)) - decode.success(#(payload, timestamp)) + use provider <- decode.field(2, decode.optional(decode.string)) + decode.success(#(payload, timestamp, provider)) }, ) |> result.map_error(fn(e) { e.message }), @@ -361,7 +368,7 @@ pub fn load_entries( use input <- result.try( unpack(row.0) |> result.replace_error("invalid saved transcript item"), ) - Ok(transcript.Entry(input, row.1)) + Ok(transcript.Entry(input, row.1, row.2)) }) }) } @@ -381,6 +388,18 @@ pub fn commit( id: String, inputs: List(types.Input), stage: String, +) -> Result(Int, String) { + commit_from(store, id, inputs, stage, None) +} + +/// Appends inputs with the provider that produced or accepted them. Provider +/// provenance lets model preparation distinguish lossless replay from transfer. +pub fn commit_from( + store: store.Store, + id: String, + inputs: List(types.Input), + stage: String, + provider: Option(String), ) -> Result(Int, String) { let timestamp = usage.now() store.query(store, fn(db) { @@ -396,12 +415,13 @@ pub fn commit( use _ <- result.try( list.try_each(inputs, fn(input) { sqlight.query( - "INSERT INTO transcript(session,payload,timestamp) VALUES(?,?,?)", + "INSERT INTO transcript(session,payload,timestamp,provider) VALUES(?,?,?,?)", db, [ sqlight.text(id), sqlight.blob(pack(input)), sqlight.int(timestamp), + sqlight.nullable(sqlight.text, provider), ], decode.dynamic, ) @@ -529,19 +549,70 @@ fn pack(input: types.Input) -> BitArray @external(erlang, "albedo_conversation", "unpack") fn unpack(bytes: BitArray) -> Result(types.Input, Nil) -pub fn set_model( +pub fn set_configuration( store: store.Store, id: String, + provider: String, model: String, + selected_protocol: types.Protocol, +) -> Result(Nil, String) { + store.query(store, fn(db) { + use _ <- result.try( + sqlight.exec("BEGIN IMMEDIATE", db) + |> result.map_error(fn(e) { e.message }), + ) + let written = { + use _ <- result.try( + sqlight.query( + "UPDATE transcript SET provider=(SELECT provider FROM sessions WHERE id=?) WHERE session=? AND provider IS NULL", + db, + [sqlight.text(id), sqlight.text(id)], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }), + ) + sqlight.query( + "UPDATE sessions SET provider=?,model=?,protocol=? WHERE id=?", + db, + [ + sqlight.text(provider), + sqlight.text(model), + sqlight.text(protocol(selected_protocol)), + sqlight.text(id), + ], + decode.dynamic, + ) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }) + } + case written { + Ok(_) -> + sqlight.exec("COMMIT", db) + |> result.replace(Nil) + |> result.map_error(fn(e) { e.message }) + Error(error) -> { + let _ = sqlight.exec("ROLLBACK", db) + Error(error) + } + } + }) +} + +/// Workspace changes are serialized by the session actor with submissions. +pub fn set_workspace( + store: store.Store, + id: String, + cwd: String, ) -> Result(Nil, String) { store.query(store, fn(db) { sqlight.query( - "UPDATE sessions SET model=? WHERE id=?", + "UPDATE sessions SET cwd=? WHERE id=?", db, - [sqlight.text(model), sqlight.text(id)], + [sqlight.text(cwd), sqlight.text(id)], decode.dynamic, ) |> result.replace(Nil) - |> result.map_error(fn(e) { e.message }) + |> result.map_error(fn(error) { error.message }) }) } diff --git a/src/albedo/daemon/projection.gleam b/src/albedo/daemon/projection.gleam new file mode 100644 index 00000000..89fad057 --- /dev/null +++ b/src/albedo/daemon/projection.gleam @@ -0,0 +1,390 @@ +//// Converts durable provider output into the portable model-input subset. +//// Raw replay remains untouched only for the provider/protocol that produced it. + +import albedo/daemon/transcript +import albedo/openai_api/types +import gleam/dict.{type Dict} +import gleam/dynamic +import gleam/dynamic/decode +import gleam/json +import gleam/list +import gleam/option.{None, Some} +import gleam/result +import gleam/string + +pub fn for_model( + newest_first: List(transcript.Entry), + provider: String, + protocol: types.Protocol, +) -> Result(List(types.Input), String) { + newest_first + |> list.reverse + |> project(provider, protocol, [], empty_chat()) +} + +fn project( + entries: List(transcript.Entry), + provider: String, + protocol: types.Protocol, + projected: List(types.Input), + pending_chat: ChatSemantic, +) -> Result(List(types.Input), String) { + case entries { + [] -> flush_chat(projected, pending_chat) + [entry, ..rest] -> + case entry.input { + types.Replay(item) -> + case + types.replay_protocol(item) == protocol + && entry.provider == Some(provider) + { + True -> { + use projected <- result.try(flush_chat(projected, pending_chat)) + project( + rest, + provider, + protocol, + [entry.input, ..projected], + empty_chat(), + ) + } + False -> + case types.replay_protocol(item), protocol { + types.Responses, types.ChatCompletions -> + case response_semantics(item) { + Ok(ResponseReasoning) -> + project(rest, provider, protocol, projected, pending_chat) + Ok(ResponseCall(call)) -> { + let ChatSemantic(text, calls) = pending_chat + project( + rest, + provider, + protocol, + projected, + ChatSemantic(text, [call, ..calls]), + ) + } + Ok(ResponseText(text)) -> { + let ChatSemantic(texts, calls) = pending_chat + project( + rest, + provider, + protocol, + projected, + ChatSemantic([text, ..texts], calls), + ) + } + Error(error) -> Error(error) + } + types.Responses, types.Responses -> { + use projected <- result.try(flush_chat( + projected, + pending_chat, + )) + use inputs <- result.try(canonical_response(item)) + project( + rest, + provider, + protocol, + list.append(list.reverse(inputs), projected), + empty_chat(), + ) + } + types.ChatCompletions, target -> { + use projected <- result.try(flush_chat( + projected, + pending_chat, + )) + use semantics <- result.try(chat_semantics(item)) + use inputs <- result.try(chat_inputs(semantics, target)) + project( + rest, + provider, + protocol, + list.append(list.reverse(inputs), projected), + empty_chat(), + ) + } + } + } + input -> { + use projected <- result.try(flush_chat(projected, pending_chat)) + project(rest, provider, protocol, [input, ..projected], empty_chat()) + } + } + } +} + +type ResponseSemantic { + ResponseReasoning + ResponseText(String) + ResponseCall(types.ToolCall) +} + +fn canonical_response( + item: types.ReplayItem, +) -> Result(List(types.Input), String) { + use semantic <- result.try(response_semantics(item)) + case semantic { + ResponseReasoning -> Ok([]) + ResponseText(text) -> Ok([types.Assistant(text)]) + ResponseCall(call) -> + response_call(call) + |> result.map(fn(item) { [types.Replay(item)] }) + } +} + +fn response_semantics( + item: types.ReplayItem, +) -> Result(ResponseSemantic, String) { + use kind <- result.try(inspect( + item, + decode.field("type", decode.string, decode.success), + "Responses output item type", + )) + case kind { + "reasoning" -> Ok(ResponseReasoning) + "message" -> response_text(item) |> result.map(ResponseText) + "function_call" -> response_call_semantics(item) |> result.map(ResponseCall) + other -> + Error( + "cannot transfer Responses output item type " + <> string.inspect(other) + <> "; its semantics are not portable", + ) + } +} + +fn response_text(item: types.ReplayItem) -> Result(String, String) { + let decoder = { + use role <- decode.field("role", decode.string) + use content <- decode.field("content", decode.list(response_part_decoder())) + case role { + "assistant" -> decode.success(string.concat(content)) + _ -> decode.failure("", "assistant response message") + } + } + inspect(item, decoder, "Responses assistant message") +} + +fn response_part_decoder() -> decode.Decoder(String) { + use kind <- decode.field("type", decode.string) + case kind { + "output_text" -> { + use text <- decode.field("text", decode.string) + decode.success(text) + } + "refusal" -> { + use text <- decode.field("refusal", decode.string) + decode.success(text) + } + other -> decode.failure("", "portable response content part, got " <> other) + } +} + +fn response_call_semantics( + item: types.ReplayItem, +) -> Result(types.ToolCall, String) { + let decoder = { + use id <- decode.field("call_id", decode.string) + use name <- decode.field("name", decode.string) + use arguments <- decode.field("arguments", decode.string) + use status <- decode.optional_field("status", "completed", decode.string) + case id != "" && name != "" && status == "completed" { + True -> decode.success(types.ToolCall(id, name, arguments)) + False -> + decode.failure( + types.ToolCall(id, name, arguments), + "completed function call with identity", + ) + } + } + inspect(item, decoder, "Responses function call") +} + +type ChatSemantic { + ChatSemantic(text: List(String), calls: List(types.ToolCall)) +} + +fn chat_semantics(item: types.ReplayItem) -> Result(ChatSemantic, String) { + let decoder = { + use fields <- decode.then(decode.dict(decode.string, decode.dynamic)) + use _ <- decode.then(reject_chat_semantics(fields)) + use content <- decode.optional_field( + "content", + None, + decode.optional(decode.string), + ) + use refusal <- decode.optional_field( + "refusal", + None, + decode.optional(decode.string), + ) + use calls <- decode.optional_field( + "tool_calls", + [], + decode.list(chat_call_decoder()), + ) + let text = + list.filter_map([content, refusal], fn(value) { + case value { + Some(text) -> Ok(text) + None -> Error(Nil) + } + }) + decode.success(ChatSemantic(text, calls)) + } + inspect(item, decoder, "Chat Completions assistant message") +} + +fn reject_chat_semantics( + fields: Dict(String, dynamic.Dynamic), +) -> decode.Decoder(Nil) { + let portable_or_metadata = [ + "role", + "content", + "refusal", + "tool_calls", + "reasoning_content", + "reasoning", + "reasoning_details", + ] + case + fields + |> dict.keys + |> list.find(fn(name) { !list.contains(portable_or_metadata, name) }) + { + Ok(name) -> + decode.failure(Nil, "known portable assistant field, got " <> name) + Error(_) -> decode.success(Nil) + } +} + +fn chat_call_decoder() -> decode.Decoder(types.ToolCall) { + use id <- decode.field("id", decode.string) + use kind <- decode.optional_field("type", "function", decode.string) + use call <- decode.field("function", { + use name <- decode.field("name", decode.string) + use arguments <- decode.field("arguments", decode.string) + decode.success(#(name, arguments)) + }) + case kind == "function" && id != "" && call.0 != "" { + True -> decode.success(types.ToolCall(id, call.0, call.1)) + False -> + decode.failure( + types.ToolCall(id, call.0, call.1), + "function tool call with identity", + ) + } +} + +fn empty_chat() -> ChatSemantic { + ChatSemantic([], []) +} + +fn chat_inputs( + semantic: ChatSemantic, + protocol: types.Protocol, +) -> Result(List(types.Input), String) { + let ChatSemantic(text, calls) = semantic + case protocol { + types.ChatCompletions -> + case calls { + [] -> Ok(list.map(text, types.Assistant)) + _ -> + chat_message(text, calls) + |> result.map(fn(item) { [types.Replay(item)] }) + } + types.Responses -> { + use tools <- result.try( + list.try_map(calls, fn(call) { + response_call(call) |> result.map(types.Replay) + }), + ) + Ok(list.append(list.map(text, types.Assistant), tools)) + } + } +} + +fn flush_chat( + projected: List(types.Input), + pending: ChatSemantic, +) -> Result(List(types.Input), String) { + let ChatSemantic(text, calls) = pending + case text, calls { + [], [] -> Ok(projected) + _, _ -> { + use item <- result.try(chat_message( + list.reverse(text), + list.reverse(calls), + )) + Ok([types.Replay(item), ..projected]) + } + } +} + +fn chat_message( + text: List(String), + calls: List(types.ToolCall), +) -> Result(types.ReplayItem, String) { + let fields = [ + #("role", json.string("assistant")), + #("content", case text { + [] -> json.null() + text -> json.string(string.concat(text)) + }), + ] + let fields = case calls { + [] -> fields + calls -> { + let tools = + list.map(calls, fn(call) { + json.object([ + #("id", json.string(call.id)), + #("type", json.string("function")), + #( + "function", + json.object([ + #("name", json.string(call.name)), + #("arguments", json.string(call.arguments)), + ]), + ), + ]) + }) + [#("tool_calls", json.array(tools, fn(tool) { tool })), ..fields] + } + } + replay(types.ChatCompletions, json.object(fields)) +} + +fn response_call(call: types.ToolCall) -> Result(types.ReplayItem, String) { + replay( + types.Responses, + json.object([ + #("type", json.string("function_call")), + #("call_id", json.string(call.id)), + #("name", json.string(call.name)), + #("arguments", json.string(call.arguments)), + #("status", json.string("completed")), + ]), + ) +} + +fn replay( + protocol: types.Protocol, + value: json.Json, +) -> Result(types.ReplayItem, String) { + json.parse(json.to_string(value), types.replay_decoder(protocol)) + |> result.map_error(fn(error) { + "could not construct portable replay item: " <> string.inspect(error) + }) +} + +fn inspect( + item: types.ReplayItem, + decoder: decode.Decoder(a), + context: String, +) -> Result(a, String) { + types.inspect_item(item, decoder) + |> result.map_error(fn(error) { context <> ": " <> string.inspect(error) }) +} diff --git a/src/albedo/daemon/server.gleam b/src/albedo/daemon/server.gleam index 16a1b2ad..a7d1ba55 100644 --- a/src/albedo/daemon/server.gleam +++ b/src/albedo/daemon/server.gleam @@ -294,6 +294,15 @@ fn info_json(info: conversation.Info) -> json.Json { ]) } +fn selection_json(selection: session.ModelSelection) -> json.Json { + json.object([ + #("ok", json.bool(True)), + #("provider", json.string(selection.provider)), + #("model", json.string(selection.model)), + #("protocol", json.string(conversation.protocol(selection.protocol))), + ]) +} + fn reply(status: Int, value: json.Json) { response.new(status) |> response.set_header("content-type", "application/json") @@ -330,7 +339,17 @@ fn route( Get, ["health"] -> reply( 200, - json.object([#("ok", json.bool(True)), #("version", json.int(2))]), + json.object([ + #("ok", json.bool(True)), + #("version", json.int(2)), + #( + "capabilities", + json.array( + ["session_provider", "session_workspace"], + json.string, + ), + ), + ]), ) Get, ["sessions"] -> reply(200, json.array(actor.call(registry, 5000, List), info_json)) @@ -377,23 +396,56 @@ fn route( } case body(req, decoder) + |> result.map_error(session.Rejected) |> result.try(fn(pair) { session.submit(worker, pair.0, pair.1) }) { Ok(_) -> reply(202, json.object([#("ok", json.bool(True))])) - Error(e) -> error(409, e) + Error(session.Rejected(e)) -> error(409, e) + Error(session.WorkspaceMissing(path)) -> + reply( + 409, + json.object([ + #("code", json.string("workspace_missing")), + #("workspace", json.string(path)), + #( + "error", + json.string("workspace not found: " <> path), + ), + ]), + ) } } - Post, "model" -> { + Post, "workspace" -> { case body( req, - decode.field("model", decode.string, decode.success), + decode.field("workspace", decode.string, decode.success), + ) + |> result.try(session.set_workspace(worker, _)) + { + Ok(info) -> reply(200, info_json(info)) + Error(e) -> error(409, e) + } + } + Post, "model" -> { + let decoder = { + use model <- decode.field("model", decode.string) + use provider <- decode.optional_field( + "provider", + None, + decode.optional(decode.string), ) - |> result.try(session.set_model(worker, _)) + decode.success(#(model, provider)) + } + case + body(req, decoder) + |> result.try(fn(selection) { + session.select_model(worker, selection.0, selection.1) + }) { - Ok(_) -> reply(200, json.object([#("ok", json.bool(True))])) + Ok(selection) -> reply(200, selection_json(selection)) Error(e) -> error(409, e) } } diff --git a/src/albedo/daemon/session.gleam b/src/albedo/daemon/session.gleam index 3339403b..45e40af8 100644 --- a/src/albedo/daemon/session.gleam +++ b/src/albedo/daemon/session.gleam @@ -3,6 +3,7 @@ import albedo/daemon/configuration import albedo/daemon/conversation import albedo/daemon/events as view +import albedo/daemon/projection import albedo/daemon/transcript import albedo/daemon/usage import albedo/harness/loop @@ -42,12 +43,29 @@ pub type Report { Report(running: Bool, kernel: Option(Int), idle_ms: Int) } +pub type ModelSelection { + ModelSelection(provider: String, model: String, protocol: types.Protocol) +} + +pub type SubmissionError { + WorkspaceMissing(String) + Rejected(String) +} + +fn submission_error(error: SubmissionError) -> String { + case error { + WorkspaceMissing(path) -> "workspace not found: " <> path + Rejected(message) -> message + } +} + pub type Message { Resume Abort(String) - Submit(String, String, Subject(Result(Nil, String))) + Submit(String, String, Subject(Result(Nil, SubmissionError))) + ChangeWorkspace(String, Subject(Result(conversation.Info, String))) Interrupt(Subject(Bool)) - ChangeModel(String, Subject(Result(Nil, String))) + ChangeModel(String, Option(String), Subject(Result(ModelSelection, String))) Status(Subject(String)) Read(Int, Subject(Page)) Publish(String, String, Subject(Bool)) @@ -112,7 +130,7 @@ pub fn start( None, home, self, - list.reverse(inputs), + inputs |> tag_unknown_provider(info.provider) |> list.reverse, latest_usage, None, 0, @@ -143,7 +161,7 @@ pub fn submit( session: Session, text: String, client_id: String, -) -> Result(Nil, String) { +) -> Result(Nil, SubmissionError) { actor.call(session, 10_000, Submit(text, client_id, _)) } @@ -199,8 +217,12 @@ fn trim( fn handle(state: State, message: Message) { // Anything a client sends counts as attention; a detached session goes quiet. let state = case message { - Submit(..) | Interrupt(..) | Status(..) | Read(..) | ChangeModel(..) -> - State(..state, last_touch: now_ms()) + Submit(..) + | Interrupt(..) + | Status(..) + | Read(..) + | ChangeModel(..) + | ChangeWorkspace(..) -> State(..state, last_touch: now_ms()) _ -> state } case message { @@ -209,29 +231,40 @@ fn handle(state: State, message: Message) { Error(error) -> actor.continue( State(..state, phase: "resting") - |> emit(view.text("error", error)), + |> emit(view.text("error", submission_error(error))), ) Ok(#(state, kernel, client)) -> { let recovered = recover_pending(state, kernel) - case - conversation.commit( - runtime.ledger(state.host), - state.info.id, - recovered, - "model", - ) - { - Error(e) -> actor.continue(emit(state, view.text("error", e))) - Ok(timestamp) -> { - let state = remember(state, recovered, timestamp) + let candidate = remember(state, recovered, 0) + case projected_inputs(candidate) { + Error(error) -> actor.continue( - start_run(state, kernel, client, model_inputs(state.history)) - |> emit(view.text( - "error", - "runtime restarted; python namespace was reset. Resuming from saved work, not replaying cells.", - )), + State(..state, phase: "resting") + |> emit(view.text("error", error)), ) - } + Ok(model_history) -> + case + conversation.commit_from( + runtime.ledger(state.host), + state.info.id, + recovered, + "model", + Some(state.info.provider), + ) + { + Error(error) -> + actor.continue(emit(state, view.text("error", error))) + Ok(timestamp) -> { + let state = remember(state, recovered, timestamp) + actor.continue( + start_run(state, kernel, client, model_history) + |> emit(view.text( + "error", + "runtime restarted; python namespace was reset. Resuming from saved work, not replaying cells.", + )), + ) + } + } } } } @@ -247,13 +280,16 @@ fn handle(state: State, message: Message) { Submit(text, client_id, reply) -> case state.run { Some(_) -> { - process.send(reply, Error("session is busy")) + process.send(reply, Error(Rejected("session is busy"))) actor.continue(state) } None -> case string.trim(text) == "" || string.byte_size(text) > 1_048_576 { True -> { - process.send(reply, Error("prompt must contain 1..1048576 bytes")) + process.send( + reply, + Error(Rejected("prompt must contain 1..1048576 bytes")), + ) actor.continue(state) } False -> @@ -267,21 +303,13 @@ fn handle(state: State, message: Message) { list.append(recover_pending(state, kernel), [ types.User(text), ]) - case - conversation.commit( - runtime.ledger(state.host), - state.info.id, - accepted, - "model", - ) - { + let candidate = remember(state, accepted, 0) + case projected_inputs(candidate) { Error(error) -> { - process.send(reply, Error(error)) + process.send(reply, Error(Rejected(error))) actor.continue(state) } - Ok(timestamp) -> { - let state = remember(state, accepted, timestamp) - let history = model_inputs(state.history) + Ok(history) -> { let model_history = case state.notice { None -> history Some(notice) -> @@ -293,19 +321,36 @@ fn handle(state: State, message: Message) { _ -> history } } - let state = - State(..state, phase: "preparing", notice: None) - let state = - start_run(state, kernel, client, model_history) - let event = - view.user( - text, - "chat", - Some(client_id), - Some(timestamp), + case + conversation.commit_from( + runtime.ledger(state.host), + state.info.id, + accepted, + "model", + Some(state.info.provider), ) - process.send(reply, Ok(Nil)) - actor.continue(emit(state, event)) + { + Error(error) -> { + process.send(reply, Error(Rejected(error))) + actor.continue(state) + } + Ok(timestamp) -> { + let state = remember(state, accepted, timestamp) + let state = + State(..state, phase: "preparing", notice: None) + let state = + start_run(state, kernel, client, model_history) + let event = + view.user( + text, + "chat", + Some(client_id), + Some(timestamp), + ) + process.send(reply, Ok(Nil)) + actor.continue(emit(state, event)) + } + } } } } @@ -328,7 +373,51 @@ fn handle(state: State, message: Message) { actor.continue(State(..state, run: Some(Run(..run, cancelled: True)))) } } - ChangeModel(model, reply) -> { + ChangeWorkspace(cwd, reply) -> { + let changed = case state.run, directory(cwd) { + Some(_), _ -> Error("session must be idle to change workspace") + None, False -> Error("workspace must be an existing absolute directory") + None, True -> + conversation.set_workspace( + runtime.ledger(state.host), + state.info.id, + cwd, + ) + } + case changed { + Error(error) -> { + process.send(reply, Error(error)) + actor.continue(state) + } + Ok(_) -> { + let state = case cwd == state.info.cwd { + True -> state + False -> { + let state = case state.kernel { + Some(kernel) -> { + let saved = + save_state_within(state, kernel, close_state_timeout) + emit( + state, + view.text("note", released_text(saved, "workspace changed")), + ) + } + None -> state + } + runtime.reset_session(state.host, state.info.id) + State( + ..state, + kernel: None, + info: conversation.Info(..state.info, cwd: cwd), + ) + } + } + process.send(reply, Ok(state.info)) + actor.continue(state) + } + } + } + ChangeModel(model, provider_name, reply) -> { case state.run == None && string.trim(model) != "" @@ -339,19 +428,48 @@ fn handle(state: State, message: Message) { actor.continue(state) } True -> { - let written = - conversation.set_model( + let selected = { + use #(provider, protocol) <- result.try(case provider_name { + None -> Ok(#(state.info.provider, state.info.protocol)) + Some(name) -> + configuration.named(state.home, name) + |> result.map(fn(configured) { + #(configured.name, configured.protocol) + }) + }) + let selection = ModelSelection(provider, model, protocol) + use _ <- result.try( + projection.for_model(state.history, provider, protocol) + |> result.replace(Nil) + |> result.map_error(fn(error) { + "cannot switch provider: " <> error + }), + ) + use _ <- result.try(conversation.set_configuration( runtime.ledger(state.host), state.info.id, + provider, model, - ) - process.send(reply, written) - case written { - Ok(_) -> + protocol, + )) + Ok(selection) + } + process.send(reply, selected) + case selected { + Ok(selection) -> actor.continue( State( ..state, - info: conversation.Info(..state.info, model: model), + history: tag_unknown_provider( + state.history, + state.info.provider, + ), + info: conversation.Info( + ..state.info, + provider: selection.provider, + model: selection.model, + protocol: selection.protocol, + ), ), ) Error(_) -> actor.continue(state) @@ -411,11 +529,12 @@ fn handle(state: State, message: Message) { Some(run) if run.id == id -> { // Completed tool results are saved even when cancellation was requested. let written = - conversation.commit( + conversation.commit_from( runtime.ledger(state.host), state.info.id, inputs, phase, + Some(state.info.provider), ) process.send(reply, written) case written { @@ -461,11 +580,12 @@ fn handle(state: State, message: Message) { _, _ -> "interrupted" } let persisted = - conversation.commit( + conversation.commit_from( runtime.ledger(state.host), state.info.id, [], stage, + Some(state.info.provider), ) let state = State(..state, run: None, phase: "resting") let state = case run.cancelled, outcome, persisted { @@ -510,7 +630,10 @@ fn handle(state: State, message: Message) { process.send(reply, True) actor.continue( State(..state, kernel: None) - |> emit(view.text("note", released_text(saved))), + |> emit(view.text( + "note", + released_text(saved, "nothing was attached"), + )), ) } _, _ -> { @@ -550,9 +673,17 @@ fn now_ms() -> Int fn prepare_submission( state: State, -) -> Result(#(State, runtime.Session, types.Client), String) { - use #(state, client) <- result.try(configured_client(state)) - use #(state, kernel) <- result.try(ensure_kernel(state)) +) -> Result(#(State, runtime.Session, types.Client), SubmissionError) { + use _ <- result.try(case directory(state.info.cwd) { + True -> Ok(Nil) + False -> Error(WorkspaceMissing(state.info.cwd)) + }) + use #(state, client) <- result.try( + configured_client(state) |> result.map_error(Rejected), + ) + use #(state, kernel) <- result.try( + ensure_kernel(state) |> result.map_error(Rejected), + ) Ok(#(state, kernel, client)) } @@ -674,10 +805,14 @@ fn restored_text(saved: python.Saved) -> String { } } -fn released_text(saved: Result(python.Saved, String)) -> String { +fn released_text( + saved: Result(python.Saved, String), + reason: String, +) -> String { + let prefix = "python kernel released: " <> reason <> "; " case saved { Ok(python.Saved([_, ..] as names, missed, engine)) -> - "python kernel released while nothing was attached; " + prefix <> int.to_string(list.length(names)) <> " variables saved to disk" <> case missed, engine { @@ -688,8 +823,7 @@ fn released_text(saved: Result(python.Saved, String)) -> String { <> " skipped (install dill to also save functions and classes)" _, _ -> ", " <> int.to_string(list.length(missed)) <> " skipped" } - _ -> - "python kernel released while nothing was attached; variables are gone, the transcript is intact" + _ -> prefix <> "variables are gone, the transcript is intact" } } @@ -710,6 +844,7 @@ fn configured_client(state: State) -> Result(#(State, types.Client), String) { "" -> State( ..state, + history: tag_unknown_provider(state.history, provider.name), info: conversation.Info(..state.info, provider: provider.name), ) _ -> state @@ -753,18 +888,41 @@ fn start_run( State(..state, run: Some(active), phase: "preparing") } -fn model_inputs(entries: List(transcript.Entry)) -> List(types.Input) { +fn raw_inputs(entries: List(transcript.Entry)) -> List(types.Input) { list.map(entries, fn(entry) { entry.input }) } +fn projected_inputs(state: State) -> Result(List(types.Input), String) { + projection.for_model(state.history, state.info.provider, state.info.protocol) + |> result.map_error(fn(error) { "cannot prepare model history: " <> error }) +} + +fn tag_unknown_provider( + entries: List(transcript.Entry), + provider: String, +) -> List(transcript.Entry) { + case provider { + "" -> entries + provider -> + list.map(entries, fn(entry) { + case entry.provider { + Some(_) -> entry + None -> transcript.Entry(..entry, provider: Some(provider)) + } + }) + } +} + fn remember(state: State, inputs: List(types.Input), timestamp: Int) -> State { let entries = - list.map(inputs, fn(input) { transcript.Entry(input, Some(timestamp)) }) + list.map(inputs, fn(input) { + transcript.Entry(input, Some(timestamp), Some(state.info.provider)) + }) State(..state, history: list.append(list.reverse(entries), state.history)) } fn recover_pending(state: State, kernel: runtime.Session) -> List(types.Input) { - let inputs = state.history |> list.reverse |> model_inputs + let inputs = state.history |> list.reverse |> raw_inputs let completed = list.filter_map(inputs, fn(input) { case input { @@ -778,6 +936,24 @@ fn recover_pending(state: State, kernel: runtime.Session) -> List(types.Input) { list.map(pending, runtime.recover(state.host, kernel, _)) } +pub fn select_model( + session: Session, + model: String, + provider: Option(String), +) -> Result(ModelSelection, String) { + actor.call(session, 5000, ChangeModel(model, provider, _)) +} + pub fn set_model(session: Session, model: String) -> Result(Nil, String) { - actor.call(session, 5000, ChangeModel(model, _)) + select_model(session, model, None) |> result.replace(Nil) +} + +@external(erlang, "albedo_daemon", "directory") +fn directory(path: String) -> Bool + +pub fn set_workspace( + session: Session, + cwd: String, +) -> Result(conversation.Info, String) { + actor.call(session, 20_000, ChangeWorkspace(cwd, _)) } diff --git a/src/albedo/daemon/transcript.gleam b/src/albedo/daemon/transcript.gleam index a8258b4c..d5e81dd1 100644 --- a/src/albedo/daemon/transcript.gleam +++ b/src/albedo/daemon/transcript.gleam @@ -4,5 +4,5 @@ import gleam/option.{type Option} /// A durable transcript item and the daemon time when it was accepted. /// Legacy rows have no timestamp; unknown time is never reconstructed. pub type Entry { - Entry(input: types.Input, timestamp: Option(Int)) + Entry(input: types.Input, timestamp: Option(Int), provider: Option(String)) } diff --git a/test/daemon/conversation_test.gleam b/test/daemon/conversation_test.gleam index f54ec5b5..f9e6e905 100644 --- a/test/daemon/conversation_test.gleam +++ b/test/daemon/conversation_test.gleam @@ -292,11 +292,14 @@ pub fn transcript_timestamps_migrate_without_invention_and_roundtrip_test() { // must not claim to know when this row was accepted. let assert Ok(_) = store.query(ledger, fn(db) { - sqlight.exec("ALTER TABLE transcript DROP COLUMN timestamp", db) + sqlight.exec( + "ALTER TABLE transcript DROP COLUMN timestamp; ALTER TABLE transcript DROP COLUMN provider", + db, + ) }) let assert Ok(_) = conversation.initialise(ledger) conversation.load_entries(ledger, "timestamped") - |> should.equal(Ok([transcript.Entry(types.User("legacy"), None)])) + |> should.equal(Ok([transcript.Entry(types.User("legacy"), None, None)])) let new_inputs = [types.User("current"), types.Assistant("answer")] let assert Ok(timestamp) = @@ -305,9 +308,9 @@ pub fn transcript_timestamps_migrate_without_invention_and_roundtrip_test() { let assert Ok(entries) = conversation.load_entries(ledger, "timestamped") entries |> should.equal([ - transcript.Entry(types.User("legacy"), None), - transcript.Entry(types.User("current"), Some(timestamp)), - transcript.Entry(types.Assistant("answer"), Some(timestamp)), + transcript.Entry(types.User("legacy"), None, None), + transcript.Entry(types.User("current"), Some(timestamp), None), + transcript.Entry(types.Assistant("answer"), Some(timestamp), None), ]) conversation.load(ledger, "timestamped") |> should.equal(Ok([types.User("legacy"), ..new_inputs])) @@ -353,3 +356,86 @@ pub fn transcript_timestamps_migrate_without_invention_and_roundtrip_test() { runtime.stop(restarted) cleanup(path) } + +pub fn provider_provenance_backfills_on_switch_and_survives_restart_test() { + let path = temporary_database() + let assert Ok(host) = runtime.start(path) + let ledger = runtime.ledger(host) + let assert Ok(_) = conversation.initialise(ledger) + let assert Ok(_) = conversation.create(ledger, session("provenance")) + let original = [types.User("old provider input"), types.Assistant("answer")] + let assert Ok(old_timestamp) = + conversation.commit(ledger, "provenance", original, "idle") + let assert Ok(before_payloads) = + store.query(ledger, fn(db) { + sqlight.query( + "SELECT payload FROM transcript WHERE session=? ORDER BY seq", + db, + [sqlight.text("provenance")], + decode.field(0, decode.bit_array, decode.success), + ) + }) + let assert Ok(_) = + conversation.set_configuration( + ledger, + "provenance", + "new-provider", + "new-model", + types.ChatCompletions, + ) + let assert Ok(after_payloads) = + store.query(ledger, fn(db) { + sqlight.query( + "SELECT payload FROM transcript WHERE session=? ORDER BY seq", + db, + [sqlight.text("provenance")], + decode.field(0, decode.bit_array, decode.success), + ) + }) + after_payloads |> should.equal(before_payloads) + let assert Ok(new_timestamp) = + conversation.commit_from( + ledger, + "provenance", + [types.User("new provider input")], + "idle", + Some("new-provider"), + ) + let expected = [ + transcript.Entry( + types.User("old provider input"), + Some(old_timestamp), + Some("provider"), + ), + transcript.Entry( + types.Assistant("answer"), + Some(old_timestamp), + Some("provider"), + ), + transcript.Entry( + types.User("new provider input"), + Some(new_timestamp), + Some("new-provider"), + ), + ] + conversation.load_entries(ledger, "provenance") |> should.equal(Ok(expected)) + conversation.load(ledger, "provenance") + |> should.equal( + Ok([ + types.User("old provider input"), + types.Assistant("answer"), + types.User("new provider input"), + ]), + ) + + runtime.stop(host) + let assert Ok(restarted) = runtime.start(path) + let ledger = runtime.ledger(restarted) + let assert Ok(_) = conversation.initialise(ledger) + conversation.load_entries(ledger, "provenance") |> should.equal(Ok(expected)) + let assert Ok([info]) = conversation.list(ledger) + #(info.provider, info.model, info.protocol) + |> should.equal(#("new-provider", "new-model", types.ChatCompletions)) + runtime.stop(restarted) + cleanup(path) +} diff --git a/test/daemon/events_test.gleam b/test/daemon/events_test.gleam index 9bf54ab4..85785851 100644 --- a/test/daemon/events_test.gleam +++ b/test/daemon/events_test.gleam @@ -48,7 +48,11 @@ pub fn snapshot_surfaces_saved_reasoning_apart_from_the_answer_test() { types.replay_decoder(types.ChatCompletions), ) let chat_events = - view.snapshot(store, [transcript.Entry(types.Replay(chat), None)], None) + view.snapshot( + store, + [transcript.Entry(types.Replay(chat), None, None)], + None, + ) types_of(chat_events) |> should.equal(["thinking", "message"]) texts(chat_events, "thinking") |> should.equal(["think"]) texts(chat_events, "message") |> should.equal(["answer"]) @@ -67,8 +71,8 @@ pub fn snapshot_surfaces_saved_reasoning_apart_from_the_answer_test() { view.snapshot( store, [ - transcript.Entry(types.Replay(reasoning), None), - transcript.Entry(types.Replay(message), None), + transcript.Entry(types.Replay(reasoning), None, None), + transcript.Entry(types.Replay(message), None, None), ], None, ) diff --git a/test/daemon/integration.py b/test/daemon/integration.py index 19bfd7d8..65f784c8 100644 --- a/test/daemon/integration.py +++ b/test/daemon/integration.py @@ -30,6 +30,7 @@ class Provider(http.server.BaseHTTPRequestHandler): "path": self.path, "authorization": self.headers.get("Authorization"), "model": request.get("model"), + "request": request, }) chat = self.path.endswith("chat/completions") inputs = request["messages" if chat else "input"] @@ -41,6 +42,10 @@ class Provider(http.server.BaseHTTPRequestHandler): code = "import asyncio, os\nfrom pathlib import Path\nPath('example.txt').write_text('hello\\n')\nsaved = Path('example.txt').read_text()\nassert 'ALBEDO_API_KEY' not in os.environ\nassert 'ALBEDO_TOKEN' not in os.environ\nawait asyncio.sleep(" + ("30" if slow else "0.4") + ")\nlen(saved)" if long_run: code = "1" + latest_user = next((item.get("content", "") for item in reversed(inputs) if item.get("role") == "user"), "") + if latest_user.startswith("workspace probe"): + done = inputs[-1].get("role") == "tool" or inputs[-1].get("type") == "function_call_output" + code = "import os\nfrom pathlib import Path\nPath('cwd-probe').write_text(os.getcwd())" arguments = json.dumps({"code": code, "timeout_ms": 60000}) self.send_response(200) self.send_header("Content-Type", "text/event-stream") @@ -84,6 +89,8 @@ def run(protocol, endpoint): home = Path(directory)/"home" workspace = Path(directory)/"workspace" workspace.mkdir() + switch_workspace = Path(directory)/"switch-workspace" + switch_workspace.mkdir() # An old database has neither provider bindings nor the new column. home.mkdir(mode=0o700) legacy_id = "legacy-fixture" @@ -105,8 +112,8 @@ def run(protocol, endpoint): def configure(active, providers): home.mkdir(parents=True, exist_ok=True) (home/"config.json").write_text(json.dumps({"active": active, "providers": providers})) - def provider(base, key, model): - return {"baseUrl": endpoint + base, "apiKey": key, "model": model, "protocol": protocol} + def provider(base, key, model, provider_protocol=protocol): + return {"baseUrl": endpoint + base, "apiKey": key, "model": model, "protocol": provider_protocol} def api(path, body=None): headers={"Authorization":"Bearer " + connection["token"], "Content-Type":"application/json"} req=urllib.request.Request(base+path, headers=headers, data=None if body is None else json.dumps(body).encode()) @@ -127,11 +134,29 @@ def run(protocol, endpoint): if line.startswith(b"data: "): page=json.loads(line[6:]) return page["events"] + def assert_projected_history(record, target_protocol, users): + request=record["request"] + items=request["messages" if target_protocol == "chat_completions" else "input"] + saved=json.dumps(items) + for user in users: + assert user in saved, (user, items) + assert "finished" in saved, items + if target_protocol == "chat_completions": + calls=[call["id"] for item in items if item.get("role") == "assistant" for call in item.get("tool_calls", [])] + outputs=[item["tool_call_id"] for item in items if item.get("role") == "tool"] + else: + calls=[item["call_id"] for item in items if item.get("type") == "function_call"] + outputs=[item["call_id"] for item in items if item.get("type") == "function_call_output"] + assert "call-1" in calls, (calls, items) + assert "call-1" in outputs, (outputs, items) try: # The daemon is useful before login: it starts and lists saved sessions. cli("sessions") connection=json.loads((home/"daemon.json").read_text()) base=f"http://127.0.0.1:{connection['port']}" + with api("/health") as response: + health=json.load(response) + assert health == {"ok": True, "version": 2, "capabilities": ["session_provider", "session_workspace"]}, health try: api("/sessions", {"workspace": str(workspace)}).close() raise AssertionError("unconfigured session creation succeeded") @@ -163,10 +188,103 @@ def run(protocol, endpoint): assert json.loads((home/"daemon.json").read_text())["pid"] == connection["pid"] # Config is re-read for each turn. The session stays on alpha even after beta becomes active. + other_protocol = "chat_completions" if protocol == "responses" else "responses" configure("beta", { "alpha": provider("/alpha/v1", "alpha-2", "ignored-new-default"), "beta": provider("/beta/v1", "beta-1", "fixture-beta"), + "gamma": provider("/gamma/v1", "gamma-1", "fixture-gamma", other_protocol), }) + + # Repair a renamed workspace without restarting daemon or losing history. + original_workspace = Path(directory)/"original-workspace" + original_workspace.mkdir() + moved_workspace = Path(directory)/"moved-workspace" + moved = json.loads(cli("new", str(original_workspace)))["session"] + with api(f"/sessions/{moved}/events", {"content":"workspace probe before move"}): + pass + ready(moved) + assert Path((original_workspace/"cwd-probe").read_text()).resolve() == original_workspace.resolve() + original_workspace.rename(moved_workspace) + before_repair = snapshot(moved) + try: + api(f"/sessions/{moved}/events", {"content":"workspace probe after move"}).close() + raise AssertionError("missing workspace accepted a turn") + except urllib.error.HTTPError as error: + failure = json.load(error) + assert error.code == 409 and failure["code"] == "workspace_missing", failure + assert failure["workspace"] == str(original_workspace), failure + for invalid in ["relative", str(original_workspace), str(moved_workspace/"cwd-probe")]: + try: + api(f"/sessions/{moved}/workspace", {"workspace":invalid}).close() + raise AssertionError("invalid replacement workspace accepted") + except urllib.error.HTTPError as error: + assert error.code == 409 + with api(f"/sessions/{moved}/workspace", {"workspace":str(moved_workspace)}) as response: + repaired = json.load(response) + assert repaired["workspace"] == str(moved_workspace), repaired + assert json.loads((home/"daemon.json").read_text())["pid"] == connection["pid"] + with api(f"/sessions/{moved}/events", {"content":"workspace probe after move"}): + pass + ready(moved) + assert Path((moved_workspace/"cwd-probe").read_text()).resolve() == moved_workspace.resolve() + assert not original_workspace.exists() + after_repair = snapshot(moved) + assert [event for event in before_repair if event.get("type") == "user"] == [event for event in after_repair if event.get("type") == "user"][:-1] + with sqlite3.connect(home/"albedo.sqlite") as db: + assert db.execute("SELECT cwd FROM sessions WHERE id=?", (moved,)).fetchone()[0] == str(moved_workspace) + + # Switch only after a complete assistant/tool conversation exists. + switched=json.loads(cli("new",str(switch_workspace)))["session"] + with api(f"/sessions/{switched}/model", {"provider":"alpha", "model":"initial-alpha"}): + pass + request_start=len(Provider.requests) + with api(f"/sessions/{switched}/events", {"content":"build switch history"}): + pass + ready(switched) + initial_requests=Provider.requests[request_start:] + assert len(initial_requests) >= 2, initial_requests + assert all("/alpha/v1/" in item["path"] for item in initial_requests), initial_requests + + # The model endpoint atomically moves that history to another protocol. + with api(f"/sessions/{switched}/model", {"provider":"gamma", "model":"chosen-gamma"}) as response: + selection=json.load(response) + assert selection == {"ok": True, "provider":"gamma", "model":"chosen-gamma", "protocol":other_protocol}, selection + # Existing model-only callers keep the current provider and protocol. + with api(f"/sessions/{switched}/model", {"model":"renamed-gamma"}) as response: + selection=json.load(response) + assert selection == {"ok": True, "provider":"gamma", "model":"renamed-gamma", "protocol":other_protocol}, selection + try: + api(f"/sessions/{switched}/model", {"provider":"unknown", "model":"wrong"}).close() + raise AssertionError("unknown provider switch succeeded") + except urllib.error.HTTPError as error: + assert error.code == 409 + request_start=len(Provider.requests) + with api(f"/sessions/{switched}/events", {"content":"after first switch"}): + pass + ready(switched) + gamma_requests=Provider.requests[request_start:] + assert gamma_requests and all("/gamma/v1/" in item["path"] for item in gamma_requests), gamma_requests + assert all(item["authorization"] == "Bearer gamma-1" for item in gamma_requests), gamma_requests + assert all(item["model"] == "renamed-gamma" for item in gamma_requests), gamma_requests + assert_projected_history(gamma_requests[0], other_protocol, ["build switch history", "after first switch"]) + + # Switching back projects the newer foreign turn while preserving the + # original provider's raw replay and every tool/result association. + with api(f"/sessions/{switched}/model", {"provider":"alpha", "model":"returned-alpha"}) as response: + selection=json.load(response) + assert selection == {"ok": True, "provider":"alpha", "model":"returned-alpha", "protocol":protocol}, selection + request_start=len(Provider.requests) + with api(f"/sessions/{switched}/events", {"content":"after second switch"}): + pass + ready(switched) + returned_requests=Provider.requests[request_start:] + assert returned_requests and all("/alpha/v1/" in item["path"] for item in returned_requests), returned_requests + assert all(item["model"] == "returned-alpha" for item in returned_requests), returned_requests + assert_projected_history(returned_requests[0], protocol, ["build switch history", "after first switch", "after second switch"]) + with api("/sessions") as response: + switched_info=next(item for item in json.load(response) if item["id"] == switched) + assert (switched_info["provider"], switched_info["model"], switched_info["protocol"]) == ("alpha", "returned-alpha", protocol), switched_info + request_start=len(Provider.requests) with api(f"/sessions/{id}/events",{"content":"write and inspect a file"}): pass @@ -215,6 +333,15 @@ def run(protocol, endpoint): with api(f"/sessions/{second}/events",{"content":"hang"}): pass time.sleep(.5) + for operation, body in [("workspace", {"workspace":str(moved_workspace)}), ("model", {"provider":"gamma", "model":"busy-rejected"})]: + try: + api(f"/sessions/{second}/{operation}", body).close() + raise AssertionError("busy session changed " + operation) + except urllib.error.HTTPError as error: + assert error.code == 409 + with api("/sessions") as response: + unchanged = next(item for item in json.load(response) if item["id"] == second) + assert unchanged["workspace"] == str(workspace) and unchanged["provider"] == "beta" with api(f"/sessions/{second}/interrupt",{}): pass ready(second) @@ -230,6 +357,7 @@ def run(protocol, endpoint): configure("alpha", { "alpha": provider("/alpha/v1", "alpha-3", "changed-alpha-default"), "beta": provider("/beta/v1", "beta-2", "changed-beta-default"), + "gamma": provider("/gamma/v1", "gamma-2", "changed-gamma-default", other_protocol), }) request_start=len(Provider.requests) os.kill(connection["pid"],9) @@ -246,17 +374,27 @@ def run(protocol, endpoint): assert all(item["model"] == "fixture-beta" for item in resumed_requests), resumed_requests with api("/sessions") as response: restored_sessions={item["id"]: item for item in json.load(response)} + assert restored_sessions[moved]["workspace"] == str(moved_workspace) assert restored_sessions[legacy_id]["provider"] == "default" assert restored_sessions[legacy_id]["title"] == "inspect the legacy workspace" assert restored_sessions[id]["provider"] == "alpha" assert restored_sessions[id]["title"] == "review the result" assert restored_sessions[third]["provider"] == "beta" assert restored_sessions[third]["title"] == "hang then recover" + assert (restored_sessions[switched]["provider"], restored_sessions[switched]["model"], restored_sessions[switched]["protocol"]) == ("alpha", "returned-alpha", protocol) + request_start=len(Provider.requests) + with api(f"/sessions/{switched}/events", {"content":"after persisted restart"}): + pass + ready(switched) + restarted_requests=Provider.requests[request_start:] + assert restarted_requests and all("/alpha/v1/" in item["path"] for item in restarted_requests), restarted_requests + assert all(item["model"] == "returned-alpha" for item in restarted_requests), restarted_requests + assert_projected_history(restarted_requests[0], protocol, ["build switch history", "after first switch", "after second switch", "after persisted restart"]) assert (workspace/"example.txt").stat().st_mtime_ns == stamp, "interrupted Python cell was replayed" recovered = snapshot(id) assert any(e.get("type")=="message" for e in recovered), "completed history lost" assert [(e["type"], e["text"], e.get("timestamp")) for e in recovered if e.get("type") in ("user", "message")] == message_times - print(protocol+": config reload, provider isolation, detach, tools, trace, interrupt, and recovery passed") + print(protocol+": provider switching, config reload, detach, tools, trace, interrupt, and recovery passed") finally: if connection: with contextlib.suppress(Exception): diff --git a/test/daemon/projection_test.gleam b/test/daemon/projection_test.gleam new file mode 100644 index 00000000..f1bebcb6 --- /dev/null +++ b/test/daemon/projection_test.gleam @@ -0,0 +1,154 @@ +import albedo/daemon/projection +import albedo/daemon/transcript +import albedo/openai_api/request +import albedo/openai_api/types +import gleam/dynamic/decode +import gleam/json +import gleam/list +import gleam/option.{None, Some} +import gleam/string +import gleam/string_tree +import gleeunit/should + +fn replay(protocol: types.Protocol, value: String) -> types.Input { + let assert Ok(item) = json.parse(value, types.replay_decoder(protocol)) + types.Replay(item) +} + +fn entry(input: types.Input, provider: String) -> transcript.Entry { + transcript.Entry(input, None, Some(provider)) +} + +fn encode(protocol: types.Protocol, newest: List(transcript.Entry)) -> String { + let assert Ok(inputs) = projection.for_model(newest, "target", protocol) + let request = types.Request("model", None, list.reverse(inputs), [], None) + let assert Ok(body) = request.encode(protocol, request) + string_tree.to_string(body) +} + +pub fn responses_span_becomes_one_valid_chat_assistant_message_test() { + let chronological = [ + entry(types.User("before"), "source"), + entry( + replay( + types.Responses, + "{\"type\":\"function_call\",\"call_id\":\"a\",\"name\":\"first\",\"arguments\":\"{\\\"n\\\":1}\",\"status\":\"completed\"}", + ), + "source", + ), + entry( + replay( + types.Responses, + "{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"between\"}]}", + ), + "source", + ), + entry( + replay( + types.Responses, + "{\"type\":\"reasoning\",\"encrypted_content\":\"opaque\"}", + ), + "source", + ), + entry( + replay( + types.Responses, + "{\"type\":\"function_call\",\"call_id\":\"b\",\"name\":\"second\",\"arguments\":\"{}\",\"status\":\"completed\"}", + ), + "source", + ), + entry(types.ToolOutput("a", "result-a"), "source"), + entry(types.ToolOutput("b", "result-b"), "source"), + entry(types.User("after"), "source"), + ] + let body = encode(types.ChatCompletions, list.reverse(chronological)) + let assert Ok(messages) = + json.parse(body, decode.at(["messages"], decode.list(decode.dynamic))) + list.map(messages, fn(message) { + decode.run(message, decode.field("role", decode.string, decode.success)) + }) + |> should.equal([ + Ok("user"), + Ok("assistant"), + Ok("tool"), + Ok("tool"), + Ok("user"), + ]) + let assert [_, assistant, first, second, _] = messages + decode.run(assistant, decode.field("content", decode.string, decode.success)) + |> should.equal(Ok("between")) + decode.run( + assistant, + decode.at( + ["tool_calls"], + decode.list(decode.field("id", decode.string, decode.success)), + ), + ) + |> should.equal(Ok(["a", "b"])) + decode.run(first, decode.field("tool_call_id", decode.string, decode.success)) + |> should.equal(Ok("a")) + decode.run( + second, + decode.field("tool_call_id", decode.string, decode.success), + ) + |> should.equal(Ok("b")) +} + +pub fn chat_text_calls_and_results_project_to_responses_in_order_test() { + let chronological = [ + entry(types.User("before"), "source"), + entry( + replay( + types.ChatCompletions, + "{\"role\":\"assistant\",\"content\":\"working\",\"reasoning_content\":\"secret\",\"tool_calls\":[{\"id\":\"a\",\"type\":\"function\",\"function\":{\"name\":\"first\",\"arguments\":\"{}\"}},{\"id\":\"b\",\"type\":\"function\",\"function\":{\"name\":\"second\",\"arguments\":\"{}\"}}]}", + ), + "source", + ), + entry(types.ToolOutput("a", "result-a"), "source"), + entry(types.ToolOutput("b", "result-b"), "source"), + ] + let body = encode(types.Responses, list.reverse(chronological)) + let assert Ok(inputs) = + json.parse(body, decode.at(["input"], decode.list(decode.dynamic))) + let assert [user, text, first, second, first_result, second_result] = inputs + decode.run(user, decode.field("content", decode.string, decode.success)) + |> should.equal(Ok("before")) + decode.run(text, decode.field("content", decode.string, decode.success)) + |> should.equal(Ok("working")) + list.map([first, second, first_result, second_result], fn(input) { + decode.run(input, decode.field("call_id", decode.string, decode.success)) + }) + |> should.equal([Ok("a"), Ok("b"), Ok("a"), Ok("b")]) + string.contains(body, "secret") |> should.be_false +} + +pub fn matching_provider_protocol_replay_is_lossless_test() { + let item = + replay( + types.Responses, + "{\"type\":\"reasoning\",\"encrypted_content\":\"opaque\",\"future\":{\"x\":1}}", + ) + let newest = [entry(item, "target")] + let body = encode(types.Responses, newest) + let assert Ok([saved]) = + json.parse(body, decode.at(["input"], decode.list(decode.dynamic))) + decode.run(saved, decode.at(["future", "x"], decode.int)) + |> should.equal(Ok(1)) + string.contains(body, "opaque") |> should.be_true +} + +pub fn unsupported_meaningful_output_rejects_projection_test() { + let newest = [ + entry( + replay( + types.Responses, + "{\"type\":\"computer_call\",\"id\":\"important\"}", + ), + "source", + ), + ] + let assert Error(error) = + projection.for_model(newest, "target", types.ChatCompletions) + string.contains(error, "computer_call") |> should.be_true + string.contains(error, "not portable") |> should.be_true +}