diff --git a/app/src/components/conversation/conversation-pane.tsx b/app/src/components/conversation/conversation-pane.tsx index 31e7faee..e3265222 100644 --- a/app/src/components/conversation/conversation-pane.tsx +++ b/app/src/components/conversation/conversation-pane.tsx @@ -82,7 +82,7 @@ export function ConversationPane({ box, session, agent: remembered, visible, onS const state = gone ? "exited" : s ? sessionState(s, stats) : undefined; const [attempt, setAttempt] = useState(0); const feed = useTranscriptFeed(box, session, s?.dir, visible && !mock && !away, attempt); - const ask = useAsk(box, session, !mock && state === "waiting", s?.state_since); + const ask = useAsk(box, session, !mock && state === "waiting", s?.state_since, s?.ask); const [answered, setAnswered] = useState<{ at?: string; key: string }>(); // A form of questions the agent asks (Claude Code's AskUserQuestion), // from its record: the chat draws it whole and the box fills it in. stuck diff --git a/app/src/components/workspace/pane.tsx b/app/src/components/workspace/pane.tsx index 50c26f6a..902666a0 100644 --- a/app/src/components/workspace/pane.tsx +++ b/app/src/components/workspace/pane.tsx @@ -228,8 +228,10 @@ export function usePaneView(pane: Leaf): "terminal" | "conversation" | undefined const s = st.boxes[c.box]?.sessions?.find((x) => x.name === c.session); return !!(s && agentOf(s)); }); + // A structured agent has no screen of its own to read: its conversation. + const structured = useStore((st) => c.kind === "terminal" && st.boxes[c.box]?.sessions?.find((x) => x.name === c.session)?.driver === "claude"); if (c.kind !== "terminal" || !labs || !runsAgent) return undefined; - return c.view ?? fallback; + return c.view ?? (structured ? "conversation" : fallback); } // ViewSwitch flips an agent's pane between its terminal and its diff --git a/app/src/lib/api.ts b/app/src/lib/api.ts index 813f0798..f314fd24 100644 --- a/app/src/lib/api.ts +++ b/app/src/lib/api.ts @@ -262,7 +262,7 @@ export const boxApi = { c.box(box, "DELETE", `locations/${encodeURIComponent(location)}/worktrees/${encodeURIComponent(worktree)}${force ? "?force=1" : ""}`), // home: in the box user's home folder, tied to no worktree, in place of a // location (boxes with the "session.home" capability). - startSession: (c: Client, box: string, req: ({ location: string } | { home: true }) & { name?: string; command?: string; agent?: string; prompt?: string; model?: string; effort?: string; title?: string }) => + startSession: (c: Client, box: string, req: ({ location: string } | { home: true }) & { name?: string; command?: string; agent?: string; prompt?: string; model?: string; effort?: string; title?: string; driver?: "terminal" | "claude"; mode?: string }) => c.box(box, "POST", "sessions", req), stopSession: (c: Client, box: string, name: string) => c.box(box, "DELETE", `sessions/${encodeURIComponent(name)}`), // Names a session's work; "" clears its title. diff --git a/app/src/lib/prefs.ts b/app/src/lib/prefs.ts index ec43cd88..8c85aeec 100644 --- a/app/src/lib/prefs.ts +++ b/app/src/lib/prefs.ts @@ -43,6 +43,13 @@ export interface Prefs { labsChosen: boolean; // Labs: how an agent's pane opens, until switched. agentView: "terminal" | "conversation"; + // Labs: new Claude Code agents run structured, driven through their + // stream by berthd rather than typed into, and open as conversations + // (boxes with "driver.claude"). + structuredClaude: boolean; + // What a structured agent's permissions do: "bypass" has berthd allow + // them (questions and plans still ask), "ask" brings each to you. + approvals: "bypass" | "ask"; // Labs: zen (⌘.): no sidebar or status bar, a switcher for a tab strip, // agents as conversations. zen: boolean; @@ -96,6 +103,8 @@ const DEFAULTS: Prefs = { labs: true, labsChosen: false, agentView: "terminal", + structuredClaude: false, + approvals: "bypass", zen: false, autoUpdateBoxes: false, chatBackground: DEFAULT_CHAT_BACKGROUND, diff --git a/app/src/lib/start-work.ts b/app/src/lib/start-work.ts index edfad43f..6bf524dc 100644 --- a/app/src/lib/start-work.ts +++ b/app/src/lib/start-work.ts @@ -11,6 +11,7 @@ import type { AgentPick, ComposerTarget } from "@/lib/composer"; import { agentLabel } from "@/lib/derive"; import { plainError } from "@/lib/errors"; import { handoff, loop, review } from "@/lib/orchestrate"; +import { usePrefs } from "@/lib/prefs"; import { worktreeSlug } from "@/lib/projects"; import { usePrompts } from "@/lib/prompts"; import { boxHasRuns, runs, scheduleRuns } from "@/lib/runs"; @@ -95,6 +96,15 @@ const fail = (title: string, err: unknown, box?: string) => { return false; }; +// structured is how a Claude agent starts when Labs runs it structured: on +// a box whose berthd drives Claude Code through its stream. +function structured(box: string, agent: string): { driver?: "claude"; mode?: string } { + const p = usePrefs.getState(); + const caps = useStore.getState().boxes[box]?.info?.capabilities ?? []; + if (agent !== "claude" || !p.labs || !p.structuredClaude || !caps.includes("driver.claude")) return {}; + return { driver: "claude", mode: p.approvals === "bypass" ? "bypassPermissions" : "default" }; +} + export async function startWork(d: StartDraft): Promise { const client = useStore.getState().client; if (!client) return false; @@ -120,7 +130,7 @@ export async function startWork(d: StartDraft): Promise { toastManager.add({ type: "success", title: `Created ${wt.name}`, description: `${wt.branch ?? ""} on ${d.box}` }); return true; } - const how = { agent: pick.agent, prompt: d.text || undefined, model: pick.model || undefined, effort: pick.effort || undefined }; + const how = { agent: pick.agent, prompt: d.text || undefined, model: pick.model || undefined, effort: pick.effort || undefined, ...structured(d.box, pick.agent) }; if (d.where === "new") { const name = d.worktree?.name || freeName(d.box, d.location, d.text ? slug(d.text) : randomName()); const res = await client.box(d.box, "POST", "tasks", { diff --git a/app/src/lib/transcript-feed.ts b/app/src/lib/transcript-feed.ts index ebbc5d4a..f7f5af4a 100644 --- a/app/src/lib/transcript-feed.ts +++ b/app/src/lib/transcript-feed.ts @@ -1,6 +1,6 @@ import { useEffect, useRef, useState } from "react"; -import { boxApi, type Client, type QueuedPrompt } from "@/lib/api"; +import { type Ask, boxApi, type Client, type QueuedPrompt } from "@/lib/api"; import type { ChatSignals } from "@/lib/chat-controls"; import { noteSignals } from "@/lib/chat-signals"; import { keyOf, offOf, useConversations } from "@/lib/conversation-store"; @@ -240,11 +240,19 @@ export function useTranscriptFeed(box: string, session: string, dir: string | un // numbered options, and the line above them as the question; form says it // shows a form of questions with steps, which no one number answers. A // menu is drawn a moment after the agent says it waits, so a screen -// without one is read again a few times. -export function useAsk(box: string, session: string, waiting: boolean, since?: string): { detail: string; choices: Choice[]; form?: boolean } | undefined { +// without one is read again a few times. A structured session has no menu +// on a screen: given, its ask's own choices answer instead. +export function useAsk(box: string, session: string, waiting: boolean, since?: string, given?: Ask): { detail: string; choices: Choice[]; form?: boolean } | undefined { const client = useStore((s) => s.client); const [ask, setAsk] = useState<{ detail: string; choices: Choice[]; form?: boolean }>(); + const offered = waiting && given?.choices?.length ? given : undefined; + const offeredKey = offered ? JSON.stringify(offered) : ""; useEffect(() => { + if (offered) { + const what = offered.input ? `${offered.tool ?? "A tool"}: ${offered.input}` : (offered.tool ?? ""); + setAsk({ detail: offered.message || what, choices: offered.choices ?? [] }); + return; + } if (!client || !waiting) { setAsk(undefined); return; @@ -281,7 +289,9 @@ export function useAsk(box: string, session: string, waiting: boolean, since?: s alive = false; window.clearTimeout(timer); }; - }, [client, box, session, waiting, since]); + // offeredKey stands for offered: a new list with the same choices + // reads the same. + }, [client, box, session, waiting, since, offeredKey]); return ask; } diff --git a/app/src/views/home/widgets/agents.tsx b/app/src/views/home/widgets/agents.tsx index c05baa83..7dea6623 100644 --- a/app/src/views/home/widgets/agents.tsx +++ b/app/src/views/home/widgets/agents.tsx @@ -131,9 +131,9 @@ function WaitingRow({ r }: { r: AgentRow }) { const s = r.session; const tool = s.ask?.tool; const question = !tool || QUESTION_TOOLS.test(tool); - // A permission's options are on the agent's screen: read them while the - // row can be seen, as the agent's own pane does. - const ask = useAsk(r.box, s.name, visible && !preview && !question, s.state_since); + // A permission's options are on the agent's screen (or, structured, on + // its ask): read them while the row can be seen, as its own pane does. + const ask = useAsk(r.box, s.name, visible && !preview && !question, s.state_since, s.ask); const choices = ask && !ask.form ? permissionChoices(ask.choices) : undefined; const allow = choices?.find((c) => c.label === "Allow"); const deny = choices?.find((c) => c.label === "Deny"); diff --git a/app/src/views/settings/labs-section.tsx b/app/src/views/settings/labs-section.tsx index adb84e1b..38077675 100644 --- a/app/src/views/settings/labs-section.tsx +++ b/app/src/views/settings/labs-section.tsx @@ -7,6 +7,8 @@ import { SettingsGroup, SettingsPage, SettingsRow } from "@/views/settings/rows" export function LabsSection() { const labs = usePrefs((p) => p.labs); const agentView = usePrefs((p) => p.agentView); + const structured = usePrefs((p) => p.structuredClaude); + const approvals = usePrefs((p) => p.approvals); return ( @@ -25,6 +27,26 @@ export function LabsSection() { /> )} + {labs && ( + + setPrefs({ structuredClaude: on })} /> + + )} + {labs && structured && ( + + setPrefs({ approvals: v })} + /> + + )} ); diff --git a/cmd/berthd/main.go b/cmd/berthd/main.go index faa3bffd..74871b21 100644 --- a/cmd/berthd/main.go +++ b/cmd/berthd/main.go @@ -357,18 +357,20 @@ func serve(b boxHome, args []string) error { Watcher: watcher, DaemonChecks: func() []doctor.Check { return daemonChecks(b, ln.Addr().String(), *listen) }, LogDir: filepath.Join(b.dir, "logs"), - Units: &box.Units{Dir: filepath.Join(b.dir, "units")}, - Turns: turns, - Hooks: hookRunner, - Flows: box.FlowsAt(userDir, b.dir), - KitsDir: filepath.Join(b.dir, "kits"), - EnvFile: filepath.Join(userDir, "env.json"), - Paused: &box.PauseStore{Path: filepath.Join(b.dir, "paused.json")}, - Secrets: b.boxSecrets(), - Socket: b.socket(), - Phone: &box.Phone{Path: filepath.Join(b.dir, "phone.json"), Addr: tailnetAddr, Log: logger}, - Guard: &box.Guard{Path: filepath.Join(userDir, "guard.json")}, - Invites: &box.Invites{Address: func() string { return pairAddress(b) }, TTL: defaultTTL}, + // Structured agents' logs (claudedriver.go). + ConversationsDir: filepath.Join(b.dir, "conversations"), + Units: &box.Units{Dir: filepath.Join(b.dir, "units")}, + Turns: turns, + Hooks: hookRunner, + Flows: box.FlowsAt(userDir, b.dir), + KitsDir: filepath.Join(b.dir, "kits"), + EnvFile: filepath.Join(userDir, "env.json"), + Paused: &box.PauseStore{Path: filepath.Join(b.dir, "paused.json")}, + Secrets: b.boxSecrets(), + Socket: b.socket(), + Phone: &box.Phone{Path: filepath.Join(b.dir, "phone.json"), Addr: tailnetAddr, Log: logger}, + Guard: &box.Guard{Path: filepath.Join(userDir, "guard.json")}, + Invites: &box.Invites{Address: func() string { return pairAddress(b) }, TTL: defaultTTL}, Update: &box.SelfUpdate{ Executable: exe, Fingerprint: id.Fingerprint().String(), diff --git a/design/structured-agents/plan.md b/design/structured-agents/plan.md index 0206e364..3f6ca42a 100644 --- a/design/structured-agents/plan.md +++ b/design/structured-agents/plan.md @@ -1,6 +1,6 @@ # Structured agents: plan -Status: 2026-10-13. Phase 0 (spike) done: go. Phase 1 (driver interface) merged. Phase 2 (agent-host + log) in review. The spike's recordings join the repo as test fixtures in phase 3, once reviewed. +Status: 2026-10-13. Phase 0 (spike) done: go. Phases 1 (driver interface) and 2 (agent-host + log) merged. Phase 3 (Claude driver, behind Labs) in review. The spike's recordings join the repo as test fixtures in phase 3, once reviewed. ## Why @@ -91,8 +91,9 @@ Exit: a go/no-go note per item; any no-go changes the phases below. interrupt, then SIGTERM, then kill the process tree (SIGKILL leaves Bash children running). - The SQLite index (conversation id, location/worktree, agent, driver, agent - session id, title, status, cost, timestamps) comes with the Claude driver - (phase 3), the first thing that has those to record. + session id, title, status, cost, timestamps) comes with resume and the + terminal hand-off (phase 4), the first thing that looks conversations up + after their session is gone. - Push to the app via the events bus (items in the event, not a "changed" ping); no stat-polling for driver agents. - Attaching to the tmux session shows a readable live render, so `berth attach` keeps working. diff --git a/docs/changelog.mdx b/docs/changelog.mdx index 6a2824c6..9a31d533 100644 --- a/docs/changelog.mdx +++ b/docs/changelog.mdx @@ -11,6 +11,16 @@ again. ## Unreleased +- **Labs: structured Claude agents.** Turn on *Structured Claude agents* + in Settings › Labs and new Claude Code agents run without their + terminal: the box drives Claude Code through its stream and keeps its own + record of the conversation, so replies stream in as they are written, + nothing is read off a screen, and an agent keeps working while your + laptop is away or the box's berthd restarts. Permissions come to the chat + as Allow / Always allow / Deny, or with *Permissions: Bypass* (the + default) the box allows them for you; questions and plans always wait for + you. They open as conversations. Needs the box upgraded. + - **Doctor no longer says berthd runs "not as a service" when it does.** `berthd install` puts your login shell's PATH in the service, and doctor compared the service with one that had no PATH, so on a Mac it always diff --git a/docs/guides/labs.mdx b/docs/guides/labs.mdx index f5a9bcb4..543e1b3e 100644 --- a/docs/guides/labs.mdx +++ b/docs/guides/labs.mdx @@ -235,6 +235,30 @@ box with the `commands` capability: `GET /v1/sessions/{name}/commands` (cached a minute, capped at 600) and `GET …/files?q=`. MCP servers' prompts aren't listed: the agent reads them from each server once it starts. +## Structured Claude agents + +With *Structured Claude agents* on in Settings › Labs, a new Claude Code +agent runs without its terminal. The box starts `claude -p` speaking +stream-json under `berthd agent-host`, which keeps the agent running in its +tmux session and writes everything that passes between them to the box's +own record of the conversation. The box drives it from there: + +- **The conversation** is read from that record, and the reply being written + streams into the chat as it is written. +- **Permissions** come to the chat with Allow, Always allow (for the rest of + the session) and Deny. *Permissions: Bypass*, the default, has the box allow + them for you instead; questions the agent asks and its plans always wait + for you. The chat's mode menu switches between bypass, ask, accept edits, + plan and auto. +- **Stop** interrupts the turn, as Esc does in its terminal. +- **Its terminal** shows what streams by; it takes no typing. + +The agent keeps working while berthd restarts or upgrades, and the record +keeps every streamed word even if the agent is killed. Slash commands that +need Claude Code's own screen (`/login`, `/resume`, `/memory`, +`/permissions`) don't run here: start a terminal agent for those. The box +needs the `driver.claude` capability (upgrade it). + ## Zen ⌘. puts everything but the agents away: no sidebar, tab strip or status diff --git a/docs/reference/app-api.mdx b/docs/reference/app-api.mdx index 137ba6b6..5ac8cdd8 100644 --- a/docs/reference/app-api.mdx +++ b/docs/reference/app-api.mdx @@ -223,7 +223,13 @@ with), `turn` (its current or last turn), `state_seq` and `fidelity` (`hooks`, `partial` or `screen`), `queued` (prompts held until it is idle) and, while it waits, `ask` (what its hooks said it asks for). `driver` says how the box reaches the agent: `terminal` (its own screen in tmux) for -every agent today; boxes that say it list `driver` in `capabilities`. `exited` is +every agent but a structured one, or `claude` (Claude Code driven through +its stream); boxes that say it list `driver` in `capabilities`. `POST +sessions` and `POST tasks` take `driver: "claude"` with `agent: "claude"`, +and `mode`, the permission mode it starts in (`bypassPermissions` has the +box allow what it asks; questions and plans still wait), on boxes that list +`driver.claude`. A waiting structured session's `ask` carries `choices` +(`{ key, label }`): send the key to answer. `exited` is true once the program has ended. `title` names the work: the first line of the prompt it started with, or of the first prompt it was sent or typed, cut to about 48 characters; `PATCH sessions/{name}` with `{ title }` renames it diff --git a/internal/agenthost/client.go b/internal/agenthost/client.go index 18f7f557..c3b65960 100644 --- a/internal/agenthost/client.go +++ b/internal/agenthost/client.go @@ -13,6 +13,9 @@ import ( // is still there to read), or never started. var ErrNoHost = errors.New("agent-host: no host is running") +// Dial connects to the host on sock: ErrNoHost when none answers. +func Dial(ctx context.Context, sock string) (net.Conn, error) { return dial(ctx, sock) } + func dial(ctx context.Context, sock string) (net.Conn, error) { var d net.Dialer c, err := d.DialContext(ctx, "unix", sock) diff --git a/internal/box/agentpaths_test.go b/internal/box/agentpaths_test.go index 0427d494..d12193e8 100644 --- a/internal/box/agentpaths_test.go +++ b/internal/box/agentpaths_test.go @@ -53,7 +53,7 @@ func TestAnNPMInstalledAgentIsOfferedAndStarts(t *testing.T) { // The session's login shell (sh -l) knows nothing of nvm. os.WriteFile(filepath.Join(home, ".profile"), []byte("PATH=/usr/bin:/bin:$PATH\n"), 0o644) ctx := context.Background() - if _, err := s.create(ctx, "agent", "", home, "claude --model opus", "claude", nil, nil); err != nil { + if _, err := s.create(ctx, "agent", "", home, "claude --model opus", "claude", nil, nil, nil); err != nil { t.Fatal(err) } deadline := time.Now().Add(10 * time.Second) diff --git a/internal/box/api.go b/internal/box/api.go index 8a077355..d5a76f5b 100644 --- a/internal/box/api.go +++ b/internal/box/api.go @@ -13,6 +13,7 @@ import ( "regexp" "strconv" "strings" + "sync" "time" "github.com/cosscom/shipyard/internal/box/runs" @@ -58,6 +59,13 @@ type Box struct { // Drivers replace a driver by kind (driver.go); nil, or a kind not in // it, is the built-in one. Drivers map[string]Driver + // ConversationsDir holds structured agents' logs, a folder each + // (claudedriver.go); AgentHost is the command that runs `berthd + // agent-host` (nil is this program's own). + ConversationsDir string + AgentHost []string + claudeOnce sync.Once + claudeDrv *claudeDriver // Hooks, when set, may refuse actions through "before:" hooks. Hooks *hooks.Runner // Flows runs the box's and its repositories' automations. @@ -658,6 +666,12 @@ type SessionRequest struct { // location: a terminal on the box, tied to no worktree. It takes a // command or a shell, never an agent preset, and no location. Home bool `json:"home,omitempty"` + // Driver, with an agent, is how berthd runs it: "terminal" (its own + // screen; the default) or "claude" (Claude Code structured, + // claudedriver.go). Mode is a structured agent's permission mode to + // start in: "bypassPermissions" has berthd allow what it asks. + Driver string `json:"driver,omitempty"` + Mode string `json:"mode,omitempty"` } func (b *Box) addSession(w http.ResponseWriter, r *http.Request) error { @@ -691,9 +705,30 @@ func (b *Box) addSession(w http.ResponseWriter, r *http.Request) error { if !ok { return badRequest("unknown agent %q", req.Agent) } + if structured, err := wantStructured(req.Driver); err != nil { + return err + } else if structured { + l, err := b.launchClaude(p, req.Model, req.Effort, req.Mode, "") + if err != nil { + return err + } + if req.Name == "" { + req.Name = defaultSessionName(req.Location, p.Command) + } + sess, err := b.startStructuredSession(r, req.Name, req.Location, dir, req.Agent, l, req.Prompt) + if err != nil { + return err + } + sess = b.titleNew(r.Context(), sess, req.Title, req.Prompt) + b.announceOpen(r, sess, req.Open) + writeJSON(w, sess) + return nil + } if req.Command, err = AgentCommandWith(p, req.Prompt, req.Model, req.Effort); err != nil { return err } + } else if req.Driver != "" && req.Driver != DriverTerminal { + return badRequest("a driver needs an agent, not a command") } else if req.Model != "" || req.Effort != "" { return badRequest("a model or an effort needs an agent, not a command") } @@ -775,6 +810,36 @@ func (b *Box) startSession(r *http.Request, name, location, dir, command, preset return sess, nil } +// wantStructured says whether a request's driver is the structured one. +func wantStructured(driver string) (bool, error) { + switch driver { + case "", DriverTerminal: + return false, nil + case DriverClaude: + return true, nil + } + return false, badRequest("unknown driver %q: terminal or claude", driver) +} + +// startStructuredSession starts a structured agent (claudedriver.go) and +// gives it its first prompt once its host listens. +func (b *Box) startStructuredSession(r *http.Request, name, location, dir, preset string, l structuredLaunch, prompt string) (Session, error) { + data := map[string]any{"name": name, "location": location, "path": dir, "command": l.command} + if err := b.before(r, "session.start", data); err != nil { + return Session{}, err + } + sess, err := b.createAgentSessionWith(r.Context(), name, location, dir, l.command, preset, l.env, l.labels) + if err != nil { + return Session{}, err + } + data["agent"], data["driver"] = agentFor(sess), DriverClaude + b.publish(r, "session.started", data) + b.setUpOnFirstOpen(origin(r), dir) + sess = b.enrich(r.Context(), []Session{sess})[0] + b.startStructured(origin(r), gateOrigin(r), sess, prompt) + return sess, nil +} + func (b *Box) removeSession(w http.ResponseWriter, r *http.Request) error { name := r.PathValue("name") if err := b.before(r, "session.stop", map[string]any{"name": name}); err != nil { @@ -1069,32 +1134,38 @@ func (b *Box) emit(w http.ResponseWriter, r *http.Request) error { if err := b.before(r, "event.emit", map[string]any{"type": req.Type}); err != nil { return err } + private := map[string]any{} + for _, k := range []string{"title", adapters.AskKey} { + if v, ok := req.Data[k]; ok { + private[k] = v + delete(req.Data, k) + } + } + b.agentEvent(r.Context(), origin(r), req.Type, req.Data, private) + writeJSON(w, map[string]bool{"ok": true}) + return nil +} + +// agentEvent publishes what an agent does, from its hooks or its driver. +// private holds what is never published: a prompt's title (adapters.Title), +// which names the session, and what a waiting agent asks for (its tool and +// a summary of its input), which goes on the turn's wait, in the ledger's +// private file, once the event has made that wait. +func (b *Box) agentEvent(ctx context.Context, origin, typ string, data, private map[string]any) { // A tool use while the agent is already working changes nothing, and // agents use tools constantly: keep them out of the journal. - // A prompt's title (adapters.Title) names the session; it is never - // published, so the journal and hooks never see it. - title, _ := req.Data["title"].(string) - delete(req.Data, "title") - // What a waiting agent asks for (its hook's tool and a summary of its - // input) is never published: it goes on the turn's wait, in the - // ledger's private file, once the event has made that wait. - ask, hasAsk := req.Data[adapters.AskKey] - delete(req.Data, adapters.AskKey) - if b.Turns != nil && b.Turns.Redundant(req.Type, req.Data) { - writeJSON(w, map[string]bool{"ok": true}) - return nil + if b.Turns != nil && b.Turns.Redundant(typ, data) { + return } - b.publish(r, req.Type, req.Data) - if title != "" && req.Type == adapters.Started && b.Turns != nil { - if name := b.Turns.SessionOf(req.Data); name != "" { - b.nameAfter(r.Context(), name, adapters.Clip(title, adapters.TitleMax)) + b.Events.Publish(events.Event{Type: typ, Box: b.Name, Origin: origin, Data: data}) + if title, _ := private["title"].(string); title != "" && typ == adapters.Started && b.Turns != nil { + if name := b.Turns.SessionOf(data); name != "" { + b.nameAfter(ctx, name, adapters.Clip(title, adapters.TitleMax)) } } - if hasAsk && b.Turns != nil && req.Type == adapters.Waiting { - b.Turns.NoteAsk(req.Data, ask) + if ask, ok := private[adapters.AskKey]; ok && b.Turns != nil && typ == adapters.Waiting { + b.Turns.NoteAsk(data, ask) } - writeJSON(w, map[string]bool{"ok": true}) - return nil } var validEventType = regexp.MustCompile(`^[a-z][a-z0-9-]{0,31}\.[a-z][a-z0-9-]{0,31}$`) diff --git a/internal/box/claudedriver.go b/internal/box/claudedriver.go new file mode 100644 index 00000000..485ebf89 --- /dev/null +++ b/internal/box/claudedriver.go @@ -0,0 +1,969 @@ +package box + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "github.com/cosscom/shipyard/internal/agenthost" + "github.com/cosscom/shipyard/internal/events" + "github.com/cosscom/shipyard/internal/integrations/adapters" + "github.com/cosscom/shipyard/internal/transcript" +) + +// The Claude driver: Claude Code without its TUI, as `claude -p` speaking +// stream-json both ways, hosted by `berthd agent-host` in the session's +// tmux pane (internal/agenthost). Prompts, interrupts, mode changes and +// answers go in as JSON lines; the conversation, the reply being written +// and every permission Claude asks for come out of the host's log. berthd's +// own hooks stay quiet in these sessions (BERTH_DRIVER): the driver says +// what the agent is doing from the stream instead. +// +// Permissions: Claude always runs in a mode of its own short of bypass, +// with --permission-prompt-tool stdio, so each permission it wants comes +// here. "Bypass" is berthd answering every one of them yes (it works where +// Claude refuses its own bypass, as root); questions and plans always wait +// for a person. + +// DriverClaude is a structured Claude Code session. +const DriverClaude = "claude" + +// The tmux options a structured session keeps. +const ( + labelDriver = "@berth_driver" + labelConv = "@berth_conv" + labelSock = "@berth_sock" + labelBypass = "@berth_bypass" +) + +// structuredModes are the modes a structured Claude session offers. +var structuredModes = []string{"default", "acceptEdits", "plan", "auto", "bypassPermissions"} + +// interactiveTools always wait for a person, bypass or not. +var interactiveTools = map[string]bool{"AskUserQuestion": true, "ExitPlanMode": true} + +// claudeStructuredFlags make claude speak stream-json both ways. +func claudeStructuredFlags(mode string) []string { + return []string{"-p", "--input-format", "stream-json", "--output-format", "stream-json", "--verbose", + "--include-partial-messages", "--replay-user-messages", "--permission-mode", mode, "--permission-prompt-tool", "stdio"} +} + +// structuredLaunch is what starts a structured Claude session. +type structuredLaunch struct { + command string + env []string + labels map[string]string +} + +// launchClaude is the command, environment and labels that start the +// preset's Claude Code under an agent host. mode is the permission mode it +// starts in ("" is default). +func (b *Box) launchClaude(p AgentPreset, model, effort, mode, resume string) (structuredLaunch, error) { + if p.ID != "claude" { + return structuredLaunch{}, badRequest("the structured driver runs Claude Code; %s runs in its terminal", p.Name) + } + if b.ConversationsDir == "" { + return structuredLaunch{}, errors.New("this box keeps no conversations") + } + switch mode { + case "": + mode = "default" + case "default", "acceptEdits", "plan", "auto", "bypassPermissions": + default: + return structuredLaunch{}, badRequest("unknown permission mode %q", mode) + } + base, err := AgentCommandWith(p, "", model, effort) + if err != nil { + return structuredLaunch{}, err + } + host := b.AgentHost + if len(host) == 0 { + exe, err := os.Executable() + if err != nil { + return structuredLaunch{}, err + } + host = []string{exe, "agent-host"} + } + id := newConvID() + dir := filepath.Join(b.ConversationsDir, id) + if err := os.MkdirAll(dir, 0o700); err != nil { + return structuredLaunch{}, err + } + sock, err := hostSocket(id) + if err != nil { + return structuredLaunch{}, err + } + claudeMode, bypass := mode, "0" + if mode == "bypassPermissions" { + claudeMode, bypass = "default", "1" + } + words := make([]string, 0, len(host)+6) + for _, w := range host { + words = append(words, shellQuote(w)) + } + words = append(words, "--dir", shellQuote(dir), "--socket", shellQuote(sock), "--", base) + for _, f := range claudeStructuredFlags(claudeMode) { + words = append(words, shellQuote(f)) + } + if resume != "" { + words = append(words, "--resume", shellQuote(resume)) + } + return structuredLaunch{ + command: strings.Join(words, " "), + env: []string{"BERTH_DRIVER=structured"}, + labels: map[string]string{labelDriver: DriverClaude, labelConv: dir, labelSock: sock, labelBypass: bypass}, + }, nil +} + +// ownedByMe says the box user owns a file. +func ownedByMe(st os.FileInfo) bool { + sys, ok := st.Sys().(*syscall.Stat_t) + return ok && int(sys.Uid) == os.Getuid() +} + +func newConvID() string { + var b [6]byte + rand.Read(b[:]) + return strconv.FormatInt(time.Now().Unix(), 36) + "-" + hex.EncodeToString(b[:]) +} + +// hostSocket is where a host listens: under /tmp, as a unix socket's path +// is limited to about 100 bytes and a state folder's can be longer. The +// folder is the box user's alone. +func hostSocket(id string) (string, error) { + dir := filepath.Join("/tmp", "berth-"+strconv.Itoa(os.Getuid())) + if err := os.MkdirAll(dir, 0o700); err != nil { + return "", err + } + if st, err := os.Lstat(dir); err != nil || !st.IsDir() || st.Mode().Perm()&0o077 != 0 || !ownedByMe(st) { + return "", fmt.Errorf("%s is not this user's own folder", dir) + } + return filepath.Join(dir, id+".sock"), nil +} + +// claudeRequest is a permission Claude asks for (can_use_tool). +type claudeRequest struct { + ID string `json:"request_id"` + Tool string `json:"tool_name"` + ToolUseID string `json:"tool_use_id"` + Input json.RawMessage `json:"input"` + Suggestions json.RawMessage `json:"permission_suggestions"` + Description string `json:"description"` +} + +// claudeConv is what berthd knows of one structured session, from its log. +type claudeConv struct { + b *Box + name string + cwd string + conv string + sock string + log string + + mu sync.Mutex + sessionID string + mode string + bypass bool + busy bool + since time.Time + draft strings.Builder + pending []claudeRequest + replies map[string]chan json.RawMessage + next int + lastSeq int64 + exited bool + changed chan struct{} + following bool + // caught is closed once the log as it was is read back: what comes + // after is new, and is said as it comes. + caught chan struct{} + caughtOnce sync.Once +} + +// claudeDriver holds the structured sessions berthd follows. +type claudeDriver struct { + b *Box + mu sync.Mutex + convs map[string]*claudeConv +} + +func (b *Box) claude() *claudeDriver { + b.claudeOnce.Do(func() { b.claudeDrv = &claudeDriver{b: b, convs: map[string]*claudeConv{}} }) + return b.claudeDrv +} + +func (d *claudeDriver) Kind() string { return DriverClaude } + +// conv is the session's, followed from its log from the first time it is +// asked for. +func (d *claudeDriver) conv(sess Session) (*claudeConv, error) { + if sess.conv == "" || sess.sock == "" { + return nil, errors.New(sess.Name + " has no structured conversation") + } + d.mu.Lock() + c := d.convs[sess.Name] + if c == nil || c.conv != sess.conv { + c = &claudeConv{b: d.b, name: sess.Name, cwd: sess.Dir, conv: sess.conv, sock: sess.sock, + log: filepath.Join(sess.conv, agenthost.LogName), replies: map[string]chan json.RawMessage{}, changed: make(chan struct{}), caught: make(chan struct{})} + d.convs[sess.Name] = c + } + d.mu.Unlock() + c.mu.Lock() + if !c.following { + c.following = true + c.bypass = sess.bypass + go c.follow() + } + c.mu.Unlock() + return c, nil +} + +// follow reads the log: what is in it already without saying anything (it +// happened before), then each new entry as it comes, saying what the agent +// does. When the host has gone, the log is all there is. +func (c *claudeConv) follow() { + agenthost.ReadLog(c.log, 0, func(e agenthost.Entry) error { + c.apply(e, false) + return nil + }) + c.reconcile() + c.caughtOnce.Do(func() { close(c.caught) }) + var noHost time.Time + for { + c.mu.Lock() + from, done := c.lastSeq, c.exited + c.mu.Unlock() + if done { + return + } + err := agenthost.Attach(context.Background(), c.sock, from, func(e agenthost.Entry) error { + c.apply(e, true) + return nil + }) + if !errors.Is(err, agenthost.ErrNoHost) { + noHost = time.Time{} + time.Sleep(100 * time.Millisecond) + continue + } + // No host: one still starting, or one gone. What it wrote last + // says which; one that died without a word is let go after a + // while, and followed again when next asked for. + agenthost.ReadLog(c.log, from, func(e agenthost.Entry) error { + c.apply(e, true) + return nil + }) + if noHost.IsZero() { + noHost = time.Now() + } + if time.Since(noHost) > hostGone { + c.mu.Lock() + c.following = false + c.mu.Unlock() + return + } + time.Sleep(200 * time.Millisecond) + } +} + +// hostGone is how long a host may not answer before berthd stops +// following its session (until it is asked about again). +const hostGone = time.Minute + +// signal wakes whoever waits on a change. Called with c.mu held. +func (c *claudeConv) signal() { + close(c.changed) + c.changed = make(chan struct{}) +} + +// streamLine is the part of a stream-json line the driver reads. +type streamLine struct { + Type string `json:"type"` + Subtype string `json:"subtype"` + SessionID string `json:"session_id"` + PermissionMode string `json:"permissionMode"` + Parent *string `json:"parent_tool_use_id"` + IsReplay bool `json:"isReplay"` + IsError bool `json:"is_error"` + Message json.RawMessage `json:"message"` + Event *struct { + Type string `json:"type"` + Delta *struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"delta"` + } `json:"event"` + RequestID string `json:"request_id"` + Request json.RawMessage `json:"request"` + Response *struct { + RequestID string `json:"request_id"` + Subtype string `json:"subtype"` + Error string `json:"error"` + Response json.RawMessage `json:"response"` + } `json:"response"` +} + +// apply reads one log entry into what the driver knows; live, it also says +// what changed. +func (c *claudeConv) apply(e agenthost.Entry, live bool) { + c.mu.Lock() + defer c.mu.Unlock() + if e.Seq <= c.lastSeq { + return + } + c.lastSeq = e.Seq + var say []func() + defer func() { + c.signal() + if live { + for _, f := range say { + go f() + } + } + }() + switch e.Dir { + case "host": + switch e.Event { + case "started": + c.busy, c.pending, c.since = false, nil, time.Time{} + say = append(say, func() { c.event(adapters.Ready, nil, nil) }) + case "exited": + c.exited, c.busy, c.pending = true, false, nil + say = append(say, func() { c.event(adapters.Exited, nil, nil) }) + } + return + case "in": + var l streamLine + if json.Unmarshal(e.Data, &l) == nil && l.Type == "control_response" && l.Response != nil { + c.drop(l.Response.RequestID) + } + return + case "out": + default: + return + } + var l streamLine + if len(e.Data) == 0 || json.Unmarshal(e.Data, &l) != nil { + return + } + if l.SessionID != "" { + c.sessionID = l.SessionID + } + helper := l.Parent != nil && *l.Parent != "" + switch l.Type { + case "system": + if l.Subtype == "init" || l.PermissionMode != "" { + if l.PermissionMode != "" { + c.mode = l.PermissionMode + } + } + if l.Subtype == "compact_boundary" { + say = append(say, c.changedConversation) + } + case "stream_event": + if helper || l.Event == nil { + return + } + switch { + case l.Event.Type == "message_start": + c.draft.Reset() + case l.Event.Type == "content_block_delta" && l.Event.Delta != nil && l.Event.Delta.Type == "text_delta": + if c.draft.Len() < maxDraft { + c.draft.WriteString(l.Event.Delta.Text) + } + } + case "assistant": + if !helper { + c.draft.Reset() + } + say = append(say, c.changedConversation) + case "user": + if !helper && l.IsReplay { + if text, ok := promptText(l.Message); ok && !c.busy { + c.busy, c.since = true, time.UnixMilli(e.At) + title := adapters.Title(text) + say = append(say, func() { c.event(adapters.Started, map[string]any{"signal": "prompt"}, map[string]any{"title": title}) }) + } + } + say = append(say, c.changedConversation) + case "control_request": + var r claudeRequest + var head struct { + Subtype string `json:"subtype"` + } + json.Unmarshal(l.Request, &head) + if head.Subtype != "can_use_tool" || json.Unmarshal(l.Request, &r) != nil { + return + } + r.ID = l.RequestID + if live && c.bypass && !interactiveTools[r.Tool] { + say = append(say, func() { c.answer(r, true, false, "") }) + return + } + c.pending = append(c.pending, r) + ask := r.ask(c.cwd) + say = append(say, func() { + c.event(adapters.Waiting, map[string]any{"reason": askReason(r)}, map[string]any{adapters.AskKey: ask}) + }) + case "control_cancel_request": + c.drop(l.RequestID) + case "control_response": + if l.Response != nil { + if ch := c.replies[l.Response.RequestID]; ch != nil { + delete(c.replies, l.Response.RequestID) + if l.Response.Subtype == "error" { + msg, _ := json.Marshal(map[string]string{"error": l.Response.Error}) + ch <- msg + } else { + ch <- l.Response.Response + } + } + } + case "result": + if helper { + return + } + c.busy, c.pending, c.since = false, nil, time.Time{} + c.draft.Reset() + data := map[string]any{} + if l.IsError { + data["status"] = "error" + if l.Subtype == "error_during_execution" { + data["status"] = "interrupted" + } + } + say = append(say, func() { c.event(adapters.Finished, data, nil) }, c.changedConversation) + } +} + +// drop forgets a permission that was answered or withdrawn. Called with +// c.mu held. +func (c *claudeConv) drop(id string) { + for i, r := range c.pending { + if r.ID == id { + c.pending = append(c.pending[:i], c.pending[i+1:]...) + return + } + } +} + +// reconcile tells the ledger what the log says now, after a restart read +// it back: a turn the agent ended, or one it still works on or waits in. +func (c *claudeConv) reconcile() { + c.mu.Lock() + busy, pending, exited, bypass := c.busy, append([]claudeRequest(nil), c.pending...), c.exited, c.bypass + c.mu.Unlock() + if exited { + return + } + if bypass { + // Asked while berthd was away: allowed now, as it would have been. + kept := pending[:0] + for _, r := range pending { + if interactiveTools[r.Tool] { + kept = append(kept, r) + continue + } + go c.answer(r, true, false, "") + } + pending = kept + } + if c.b.Turns == nil { + return + } + st, _ := c.b.Turns.State(c.name) + switch { + case len(pending) > 0 && st.State != "waiting": + c.event(adapters.Waiting, map[string]any{"reason": askReason(pending[0])}, map[string]any{adapters.AskKey: pending[0].ask(c.cwd)}) + case busy && len(pending) == 0 && st.State != "running": + c.event(adapters.Started, map[string]any{"signal": "tool"}, nil) + case !busy && (st.State == "running" || st.State == "waiting"): + c.event(adapters.Finished, nil, nil) + } +} + +// promptText is a person's prompt in a user message: text, not a tool's +// result or Claude Code's own note. +func promptText(raw json.RawMessage) (string, bool) { + var m struct { + Content json.RawMessage `json:"content"` + } + if json.Unmarshal(raw, &m) != nil { + return "", false + } + var s string + if json.Unmarshal(m.Content, &s) == nil { + return s, s != "" && !strings.HasPrefix(s, " 0 { + resp["updatedPermissions"] = sessionOnly(r.Suggestions) + } + } + return c.respond(r, resp) +} + +func (c *claudeConv) respond(r claudeRequest, resp map[string]any) error { + err := c.write(context.Background(), map[string]any{"type": "control_response", "response": map[string]any{ + "subtype": "success", "request_id": r.ID, "response": resp, + }}) + if err == nil { + // The tool runs, or the agent goes on without it: working again. + c.event(adapters.Started, map[string]any{"signal": "tool"}, nil) + } + return err +} + +// sessionOnly keeps "don't ask again" to this session: Claude suggests +// rules for the project's settings file. +func sessionOnly(raw json.RawMessage) json.RawMessage { + var sug []map[string]any + if json.Unmarshal(raw, &sug) != nil { + return raw + } + for _, s := range sug { + s["destination"] = "session" + } + b, _ := json.Marshal(sug) + return b +} + +// first is the oldest permission still waiting. +func (c *claudeConv) first() (claudeRequest, bool) { + c.mu.Lock() + defer c.mu.Unlock() + if len(c.pending) == 0 { + return claudeRequest{}, false + } + return c.pending[0], true +} + +// planChoices answer Claude's plan; permissionChoices any other permission. +var ( + permissionChoices = []AskChoice{{Key: "1", Label: "Yes"}, {Key: "2", Label: "Yes, and don't ask again this session"}, {Key: "3", Label: "No"}} + planChoices = []AskChoice{{Key: "1", Label: "Yes, and auto-accept edits"}, {Key: "2", Label: "Yes, and approve each edit"}, {Key: "3", Label: "No, keep planning"}} +) + +// AskChoice is one answer a waiting agent offers, by the key that picks it. +type AskChoice struct { + Key string `json:"key"` + Label string `json:"label"` +} + +// choices are the answers the oldest permission offers, by key. +func (c *claudeConv) choices() []AskChoice { + r, ok := c.first() + switch { + case !ok, r.Tool == "AskUserQuestion": + return nil + case r.Tool == "ExitPlanMode": + return planChoices + } + return permissionChoices +} + +// pick answers the oldest permission with the choice its key names. +func (c *claudeConv) pick(ctx context.Context, key string) error { + r, ok := c.first() + if !ok { + return httpError{http.StatusConflict, "Claude isn't asking for anything"} + } + if r.Tool == "AskUserQuestion" { + return httpError{http.StatusConflict, "Claude asked questions: answer them in the chat"} + } + if r.Tool == "ExitPlanMode" { + switch key { + case "1", "2": + mode := "acceptEdits" + if key == "2" { + mode = "default" + } + return c.respond(r, map[string]any{"behavior": "allow", "updatedInput": json.RawMessage(r.Input), + "updatedPermissions": []map[string]any{{"type": "setMode", "mode": mode, "destination": "session"}}}) + case "3": + return c.answer(r, false, false, "The person wants to keep planning.") + } + return badRequest("pick 1, 2 or 3") + } + switch key { + case "1", "y", "Y": + return c.answer(r, true, false, "") + case "2": + return c.answer(r, true, true, "") + case "3", "n", "N": + return c.answer(r, false, false, "") + } + return badRequest("pick 1, 2 or 3") +} + +func (d *claudeDriver) Send(ctx context.Context, sess Session, text string, enter bool) error { + c, err := d.conv(sess) + if err != nil { + return err + } + if r, waiting := c.first(); waiting { + if isKey(text) && !enter { + return c.pick(ctx, text) + } + if strings.TrimSpace(text) != "" { + // Typed at a permission (sent with force): no, and this instead. + return c.answer(r, false, false, "The person said no, and: "+text) + } + } + if !enter { + return badRequest("a structured session takes whole messages") + } + if strings.TrimSpace(text) == "" { + return nil + } + return c.write(ctx, map[string]any{"type": "user", "message": map[string]any{"role": "user", "content": text}}) +} + +func (d *claudeDriver) Interrupt(ctx context.Context, sess Session) (bool, error) { + c, err := d.conv(sess) + if err != nil { + return false, err + } + if _, err := c.control(ctx, map[string]any{"subtype": "interrupt"}); err != nil { + return false, err + } + deadline := time.After(4 * time.Second) + for { + c.mu.Lock() + busy, changed := c.busy, c.changed + c.mu.Unlock() + if !busy { + return true, nil + } + select { + case <-changed: + case <-deadline: + return false, nil + case <-ctx.Done(): + return false, ctx.Err() + } + } +} + +func (d *claudeDriver) Controls(ctx context.Context, sess Session) (Controls, error) { + c, err := d.conv(sess) + if err != nil { + return Controls{Agent: "claude"}, err + } + c.mu.Lock() + defer c.mu.Unlock() + mode := firstNonEmpty(c.mode, "default") + if c.bypass { + mode = "bypassPermissions" + } + return Controls{Agent: "claude", Mode: mode, Modes: structuredModes}, nil +} + +func (d *claudeDriver) SetMode(ctx context.Context, sess Session, mode string) (string, error) { + c, err := d.conv(sess) + if err != nil { + return "", err + } + known := false + for _, m := range structuredModes { + known = known || m == mode + } + if !known { + return "", badRequest("unknown permission mode %q", mode) + } + bypass := mode == "bypassPermissions" + claudeMode := mode + if bypass { + claudeMode = "default" + } + c.mu.Lock() + current := firstNonEmpty(c.mode, "default") + c.mu.Unlock() + if current != claudeMode { + if _, err := c.control(ctx, map[string]any{"subtype": "set_permission_mode", "mode": claudeMode}); err != nil { + return "", err + } + c.mu.Lock() + c.mode = claudeMode + c.mu.Unlock() + } + c.mu.Lock() + c.bypass = bypass + var allow []claudeRequest + if bypass { + for _, r := range c.pending { + if !interactiveTools[r.Tool] { + allow = append(allow, r) + } + } + } + c.mu.Unlock() + v := "0" + if bypass { + v = "1" + } + d.b.Sessions.tmux(ctx, "set-option", "-t", "="+sess.Name+":", labelBypass, v) + for _, r := range allow { + c.answer(r, true, false, "") + } + return mode, nil +} + +func (d *claudeDriver) Answer(ctx context.Context, sess Session, req AnswerRequest) ([]string, error) { + c, err := d.conv(sess) + if err != nil { + return nil, err + } + c.mu.Lock() + var r claudeRequest + found := false + for _, p := range c.pending { + if p.Tool == "AskUserQuestion" && p.ToolUseID == req.Tool { + r, found = p, true + } + } + c.mu.Unlock() + if !found { + return nil, httpError{http.StatusConflict, "Claude isn't waiting for those answers"} + } + var in struct { + Questions []struct { + Question string `json:"question"` + Header string `json:"header"` + MultiSelect bool `json:"multiSelect"` + Options []struct { + Label string `json:"label"` + Description string `json:"description"` + } `json:"options"` + } `json:"questions"` + } + if err := json.Unmarshal(r.Input, &in); err != nil { + return nil, err + } + qs := make([]transcript.Question, len(in.Questions)) + for i, q := range in.Questions { + qs[i] = transcript.Question{Question: q.Question, Header: q.Header, Multi: q.MultiSelect} + for _, o := range q.Options { + qs[i].Options = append(qs[i].Options, transcript.Option{Label: o.Label, Description: o.Description}) + } + } + shown, err := checkAnswers(qs, req.Answers) + if err != nil { + return nil, err + } + answers := map[string]string{} + for i, q := range qs { + answers[q.Question] = shown[i] + } + var input map[string]any + json.Unmarshal(r.Input, &input) + if input == nil { + input = map[string]any{} + } + input["answers"] = answers + if err := c.respond(r, map[string]any{"behavior": "allow", "updatedInput": input}); err != nil { + return nil, err + } + return shown, nil +} + +func (d *claudeDriver) Conversation(ctx context.Context, sess Session, _ []Session, since int, gen string) (transcript.Result, error) { + if _, err := d.conv(sess); err != nil { + return transcript.Result{Source: "none", Items: []transcript.Item{}, Crew: []transcript.CrewMember{}, Reason: err.Error()}, nil + } + return reader().Follow(transcript.SourceClaudeHost, filepath.Join(sess.conv, agenthost.LogName), sess.Dir, max(since, 0), gen) +} + +func (d *claudeDriver) Record(ctx context.Context, sess Session, _ []Session) (string, string) { + return transcript.SourceClaudeHost, filepath.Join(sess.conv, agenthost.LogName) +} + +// Draft is the reply Claude is streaming, and how long it has worked. +func (d *claudeDriver) Draft(ctx context.Context, sess Session) (Draft, error) { + c, err := d.conv(sess) + if err != nil { + return Draft{Agent: "claude"}, nil + } + c.mu.Lock() + defer c.mu.Unlock() + out := Draft{Agent: "claude", Text: c.draft.String()} + if c.busy && len(c.pending) == 0 { + out.Status = &DraftStatus{Word: "Working…"} + if !c.since.IsZero() { + out.Status.Elapsed = time.Since(c.since).Round(time.Second).String() + } + } + return out, nil +} + +// askChoices are the answers a waiting structured session offers. +func (d *claudeDriver) askChoices(sess Session) []AskChoice { + c, err := d.conv(sess) + if err != nil { + return nil + } + return c.choices() +} + +// startStructured gives a new structured session its first prompt once its +// host listens. +func (b *Box) startStructured(origin, from string, sess Session, prompt string) { + go func() { + deadline := time.Now().Add(30 * time.Second) + for time.Now().Before(deadline) { + if c, err := agenthost.Dial(context.Background(), sess.sock); err == nil { + c.Close() + break + } + time.Sleep(100 * time.Millisecond) + } + // Followed from its start, so its first turn is said as it goes. + b.claude().conv(sess) + if strings.TrimSpace(prompt) == "" { + return + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + b.sendPrompt(ctx, sess.Name, SendRequest{Text: prompt, When: "now"}, origin, from) + }() +} diff --git a/internal/box/claudedriver_test.go b/internal/box/claudedriver_test.go new file mode 100644 index 00000000..e017c265 --- /dev/null +++ b/internal/box/claudedriver_test.go @@ -0,0 +1,343 @@ +package box + +import ( + "encoding/json" + "os" + "os/exec" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/cosscom/shipyard/internal/transcript" +) + +// fakeStreamClaude stands in for `claude -p` speaking stream-json, with the +// lines the spike recorded (design/structured-agents): it echoes each +// prompt, streams and writes a reply, asks for a permission to "run the +// tests", asks a question to "ask me", works until interrupted on "work", +// and answers control requests. +const fakeStreamClaude = `#!/usr/bin/env python3 +import json, sys +S = "00000000-0000-4000-8000-0000000000c1" +mode = sys.argv[sys.argv.index("--permission-mode") + 1] +n = [0] +def out(d): + d.setdefault("session_id", S) + print(json.dumps(d), flush=True) +def uid(): + n[0] += 1 + return "00000000-0000-4000-8000-%012d" % n[0] +def say(text): + out({"type": "stream_event", "event": {"type": "message_start"}, "parent_tool_use_id": None}) + out({"type": "stream_event", "event": {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}}, "parent_tool_use_id": None}) + out({"type": "assistant", "message": {"id": "msg_" + uid(), "role": "assistant", "content": [{"type": "text", "text": text}]}, "parent_tool_use_id": None, "uuid": uid()}) +def tool(name, inp): + t = "toolu_" + uid()[-8:] + out({"type": "assistant", "message": {"id": "msg_" + uid(), "role": "assistant", "content": [{"type": "tool_use", "id": t, "name": name, "input": inp}]}, "parent_tool_use_id": None, "uuid": uid()}) + return t +def result(sub="success"): + out({"type": "result", "subtype": sub, "is_error": sub != "success", "result": "", "session_id": S}) +def control_reply(req): + r = req["request"] + if r["subtype"] == "set_permission_mode": + global mode + mode = r["mode"] + out({"type": "control_response", "response": {"subtype": "success", "request_id": req["request_id"], "response": {"mode": mode}}}) + out({"type": "system", "subtype": "status", "status": None, "permissionMode": mode}) + else: + out({"type": "control_response", "response": {"subtype": "success", "request_id": req["request_id"], "response": {}}}) +def wait_for(rid): + for line in sys.stdin: + m = json.loads(line) + if m["type"] == "control_response" and m["response"]["request_id"] == rid: + return m["response"]["response"] + if m["type"] == "control_request": + control_reply(m) + if m["request"]["subtype"] == "interrupt": + return None +first = True +for line in sys.stdin: + m = json.loads(line) + if m["type"] == "control_request": + control_reply(m) + continue + text = m["message"]["content"] + out({"type": "user", "message": m["message"], "isReplay": True, "parent_tool_use_id": None, "uuid": uid()}) + if first: + out({"type": "system", "subtype": "init", "cwd": "/w", "model": "claude-fake", "permissionMode": mode, "uuid": uid()}) + first = False + if text == "run the tests": + t = tool("Bash", {"command": "pnpm test", "description": "Run the tests"}) + rid = "req_" + uid()[-6:] + out({"type": "control_request", "request_id": rid, "request": {"subtype": "can_use_tool", "tool_name": "Bash", "tool_use_id": t, "input": {"command": "pnpm test", "description": "Run the tests"}, "permission_suggestions": [{"type": "addRules", "rules": [{"toolName": "Bash", "ruleContent": "pnpm test"}], "behavior": "allow", "destination": "localSettings"}]}}) + r = wait_for(rid) + ok = r and r.get("behavior") == "allow" + out({"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": t, "content": "12 passed" if ok else "denied", "is_error": not ok}]}, "parent_tool_use_id": None, "uuid": uid()}) + say("The tests pass." if ok else "I didn't run them.") + result() + elif text == "ask me": + q = {"question": "Tea or coffee?", "header": "Drink", "options": [{"label": "Tea", "description": ""}, {"label": "Coffee", "description": ""}], "multiSelect": False} + t = tool("AskUserQuestion", {"questions": [q]}) + rid = "req_" + uid()[-6:] + out({"type": "control_request", "request_id": rid, "request": {"subtype": "can_use_tool", "tool_name": "AskUserQuestion", "tool_use_id": t, "requires_user_interaction": True, "input": {"questions": [q]}}}) + r = wait_for(rid) + a = r["updatedInput"]["answers"]["Tea or coffee?"] + out({"type": "user", "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": t, "content": "Your questions have been answered: \"Tea or coffee?\"=\"%s\"." % a}]}, "parent_tool_use_id": None, "uuid": uid()}) + say("You chose " + a + ".") + result() + elif text == "work": + out({"type": "stream_event", "event": {"type": "message_start"}, "parent_tool_use_id": None}) + out({"type": "stream_event", "event": {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "Working through it"}}, "parent_tool_use_id": None}) + wait_for("never") + result("error_during_execution") + else: + say("echo: " + text) + result() +` + +func TestAStructuredClaudeSessionIsDrivenThroughItsStream(t *testing.T) { + if _, err := exec.LookPath("python3"); err != nil { + t.Skip("the stand-in Claude needs python3") + } + bin := t.TempDir() + os.WriteFile(filepath.Join(bin, "claude"), []byte(fakeStreamClaude), 0o755) + t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) + exe, err := os.Executable() + if err != nil { + t.Fatal(err) + } + states := &Turns{} + var box *Box + c, bus := servedBox(t, func(b *Box) { + b.Turns = states + b.ConversationsDir = t.TempDir() + b.AgentHost = []string{"env", "BERTH_TEST_AGENT_HOST=1", exe} + box = b + }) + states.Attach(bus) + repo := gitRepo(t) + call(t, c, "POST", "/v1/locations", "", map[string]string{"name": "acme", "path": repo}, nil) + + var sess Session + if status := call(t, c, "POST", "/v1/sessions", "", map[string]any{"location": "acme", "agent": "claude", "driver": "claude", "prompt": "hello there"}, &sess); status != 200 { + t.Fatalf("start: %d", status) + } + t.Cleanup(func() { call(t, c, "DELETE", "/v1/sessions/"+sess.Name, "", nil, nil) }) + if sess.Driver != DriverClaude { + t.Fatalf("driver = %q", sess.Driver) + } + name := sess.Name + + // read waits for the conversation to hold what ok looks for. + read := func(ok func([]transcript.Item) bool) []transcript.Item { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for { + var res transcript.Result + call(t, c, "GET", "/v1/sessions/"+name+"/transcript?since=0", "", nil, &res) + if ok(res.Items) { + return res.Items + } + if time.Now().After(deadline) { + t.Fatalf("the conversation never showed it: %+v (%s)", res.Items, res.Reason) + } + time.Sleep(50 * time.Millisecond) + } + } + has := func(kind, text string) func([]transcript.Item) bool { + return func(items []transcript.Item) bool { + for _, it := range items { + if it.Kind == kind && strings.Contains(it.Text, text) { + return true + } + } + return false + } + } + state := func(want string) SessionState { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + st, _ := states.State(name) + if st.State == want { + return st + } + if time.Now().After(deadline) { + t.Fatalf("the agent never became %s: %+v", want, st) + } + time.Sleep(50 * time.Millisecond) + } + } + + // The first prompt goes in once the host listens, and the turn it + // starts ends, all from the stream: no hook ran. + read(has("text", "echo: hello there")) + read(has("user", "hello there")) + if st := state("finished"); st.AgentSessionID != "00000000-0000-4000-8000-0000000000c1" { + t.Fatalf("the agent's own session id = %q", st.AgentSessionID) + } + + // A permission waits for the person, with choices to answer it by. + call(t, c, "POST", "/v1/sessions/"+name+"/send", "", map[string]any{"text": "run the tests", "when": "now"}, nil) + if st := state("waiting"); st.Ask == nil || st.Ask.Tool != "Bash" || st.Ask.Input != "pnpm test" { + t.Fatalf("ask = %+v", st.Ask) + } + var list []Session + call(t, c, "GET", "/v1/sessions", "", nil, &list) + if len(list) != 1 || list[0].Ask == nil || len(list[0].Ask.Choices) != 3 || list[0].Ask.Choices[0].Label != "Yes" { + t.Fatalf("the session list doesn't offer the choices: %+v", list) + } + call(t, c, "POST", "/v1/sessions/"+name+"/send", "", map[string]any{"text": "1", "enter": false, "when": "now", "force": true}, nil) + read(has("text", "The tests pass.")) + state("finished") + + // Bypass: berthd says yes for the person. + var mode map[string]string + if status := call(t, c, "POST", "/v1/sessions/"+name+"/mode", "", map[string]string{"mode": "bypassPermissions"}, &mode); status != 200 || mode["mode"] != "bypassPermissions" { + t.Fatalf("mode: %d %v", status, mode) + } + var controls Controls + call(t, c, "GET", "/v1/sessions/"+name+"/controls", "", nil, &controls) + if controls.Mode != "bypassPermissions" { + t.Fatalf("controls = %+v", controls) + } + call(t, c, "POST", "/v1/sessions/"+name+"/send", "", map[string]any{"text": "run the tests", "when": "now"}, nil) + items := read(func(items []transcript.Item) bool { + n := 0 + for _, it := range items { + if it.Kind == "text" && it.Text == "The tests pass." { + n++ + } + } + return n == 2 + }) + if len(items) == 0 { + t.Fatal("no items") + } + + // A question waits even in bypass, and is answered from the chat. + call(t, c, "POST", "/v1/sessions/"+name+"/send", "", map[string]any{"text": "ask me", "when": "now"}, nil) + state("waiting") + q := read(func(items []transcript.Item) bool { + for _, it := range items { + if it.Kind == "question" { + return true + } + } + return false + }) + var tool string + for _, it := range q { + if it.Kind == "question" { + tool = it.Tool + } + } + var ans AnswerResult + if status := call(t, c, "POST", "/v1/sessions/"+name+"/answer", "", map[string]any{"tool": tool, "answers": []any{map[string]any{"picks": []string{"Coffee"}}}}, &ans); status != 200 || len(ans.Answered) != 1 || ans.Answered[0] != "Coffee" { + t.Fatalf("answer: %d %+v", status, ans) + } + read(has("text", "You chose Coffee.")) + state("finished") + + // The reply being written shows as a draft, and an interrupt stops it. + call(t, c, "POST", "/v1/sessions/"+name+"/send", "", map[string]any{"text": "work", "when": "now"}, nil) + deadline := time.Now().Add(10 * time.Second) + for { + var d Draft + call(t, c, "GET", "/v1/sessions/"+name+"/draft", "", nil, &d) + if d.Text == "Working through it" && d.Status != nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("draft = %+v", d) + } + time.Sleep(50 * time.Millisecond) + } + var stopped map[string]bool + call(t, c, "POST", "/v1/sessions/"+name+"/interrupt", "", nil, &stopped) + if !stopped["stopped"] { + t.Fatalf("interrupt: %v", stopped) + } + state("finished") + + // berthd starting again reads the log back, and goes on driving it. + live, err := box.Sessions.Get(t.Context(), name) + if err != nil || live.conv == "" || live.sock == "" { + t.Fatalf("the session lost its labels: %+v, %v", live, err) + } + again := &claudeDriver{b: box, convs: map[string]*claudeConv{}} + if err := again.Send(t.Context(), live, "still there?", true); err != nil { + t.Fatal(err) + } + read(has("text", "echo: still there?")) + conv, _ := again.conv(live) + conv.mu.Lock() + sid := conv.sessionID + conv.mu.Unlock() + if sid != "00000000-0000-4000-8000-0000000000c1" { + t.Fatalf("read back, the session id is %q", sid) + } + + // Its log is berthd's own record of all of it, the streamed words too. + raw, _ := os.ReadFile(filepath.Join(live.conv, "events.jsonl")) + if !json.Valid([]byte(strings.Split(string(raw), "\n")[0])) || !strings.Contains(string(raw), `"Working through it"`) { + t.Fatalf("the log doesn't hold the stream") + } +} + +// TestALiveClaudeSessionWorks drives the real Claude Code (haiku): run it +// with BERTH_LIVE_CLAUDE=1 on a machine logged in to Claude. +func TestALiveClaudeSessionWorks(t *testing.T) { + if os.Getenv("BERTH_LIVE_CLAUDE") != "1" { + t.Skip("set BERTH_LIVE_CLAUDE=1 to drive the real Claude Code") + } + if _, err := exec.LookPath("claude"); err != nil { + t.Skip("no claude on PATH") + } + exe, _ := os.Executable() + states := &Turns{} + c, bus := servedBox(t, func(b *Box) { + b.Turns = states + b.ConversationsDir = t.TempDir() + b.AgentHost = []string{"env", "BERTH_TEST_AGENT_HOST=1", exe} + }) + states.Attach(bus) + repo := gitRepo(t) + call(t, c, "POST", "/v1/locations", "", map[string]string{"name": "acme", "path": repo}, nil) + var sess Session + if status := call(t, c, "POST", "/v1/sessions", "", map[string]any{"location": "acme", "agent": "claude", "driver": "claude", "model": "haiku", "mode": "bypassPermissions", "prompt": "Reply with just the word: harbour"}, &sess); status != 200 { + t.Fatalf("start: %d", status) + } + t.Cleanup(func() { call(t, c, "DELETE", "/v1/sessions/"+sess.Name, "", nil, nil) }) + wait := func(what string, ok func([]transcript.Item, SessionState) bool) { + t.Helper() + deadline := time.Now().Add(90 * time.Second) + for { + var res transcript.Result + call(t, c, "GET", "/v1/sessions/"+sess.Name+"/transcript?since=0", "", nil, &res) + st, _ := states.State(sess.Name) + if ok(res.Items, st) { + return + } + if time.Now().After(deadline) { + t.Fatalf("%s never happened: %+v / %+v", what, res.Items, st) + } + time.Sleep(250 * time.Millisecond) + } + } + said := func(word string) func([]transcript.Item, SessionState) bool { + return func(items []transcript.Item, st SessionState) bool { + for _, it := range items { + if it.Kind == "text" && strings.Contains(strings.ToLower(it.Text), word) { + return st.State == "finished" + } + } + return false + } + } + wait("the first reply", said("harbour")) + // Bypass: the box allows the command it runs. + call(t, c, "POST", "/v1/sessions/"+sess.Name+"/send", "", map[string]any{"text": "Run `echo shipyard-live` with the Bash tool, then reply with just its output.", "when": "now"}, nil) + wait("the command's output", said("shipyard-live")) +} diff --git a/internal/box/draft.go b/internal/box/draft.go index 56d5f300..d21f0da9 100644 --- a/internal/box/draft.go +++ b/internal/box/draft.go @@ -53,6 +53,15 @@ func (b *Box) draft(w http.ResponseWriter, r *http.Request) error { if err != nil && sess.Name == "" { return err } + if d := b.driverFor(sess); d.Kind() != DriverTerminal { + // A structured session's reply streams to berthd, not its screen. + out, err := d.Draft(r.Context(), sess) + if err != nil { + return err + } + writeJSON(w, out) + return nil + } agent := firstNonEmpty(sess.Preset, firstNonEmpty(sess.Agent, agentOf(sess.Command))) out := Draft{Agent: agent} if agent == "claude" && sess.Service == "" && !sess.Exited { diff --git a/internal/box/driver.go b/internal/box/driver.go index 74e00ab8..f0e21fde 100644 --- a/internal/box/driver.go +++ b/internal/box/driver.go @@ -43,9 +43,15 @@ type Driver interface { // Conversation is the chat's view of the session from item since. // all is the box's sessions when the caller has just listed them. Conversation(ctx context.Context, sess Session, all []Session, since int, gen string) (transcript.Result, error) + // Record is the file the conversation is read from, and how: its + // older pages are read from it too. An empty path is none yet. + Record(ctx context.Context, sess Session, all []Session) (source, path string) + // Draft is the reply the agent is writing, before its record has it. + Draft(ctx context.Context, sess Session) (Draft, error) } -// DriverTerminal is the driver every session has today. +// DriverTerminal is a session's agent in its own terminal: every session +// but a structured one (DriverClaude). const DriverTerminal = "terminal" // driverFor is the driver for a session's agent. @@ -54,6 +60,9 @@ func (b *Box) driverFor(sess Session) Driver { if d := b.Drivers[kind]; d != nil { return d } + if kind == DriverClaude { + return b.claude() + } return terminalDriver{b} } @@ -197,6 +206,32 @@ func (t terminalDriver) Answer(ctx context.Context, sess Session, req AnswerRequ return shown, nil } +// Record is the agent's own transcript file. +func (t terminalDriver) Record(ctx context.Context, sess Session, all []Session) (string, string) { + agent, path, _ := t.b.transcriptFile(ctx, sess, all) + if agent != "claude" && agent != "codex" { + return "", "" + } + return agent, path +} + +// Draft reads the reply Claude Code is drawing from its styled screen +// (draft.go). Other agents aren't drafted. +func (t terminalDriver) Draft(ctx context.Context, sess Session) (Draft, error) { + agent := firstNonEmpty(sess.Preset, firstNonEmpty(sess.Agent, agentOf(sess.Command))) + out := Draft{Agent: agent} + if agent != "claude" || sess.Service != "" || sess.Exited { + return out, nil + } + _, raw, err := t.b.Sessions.getAndCapture(ctx, sess.Name, "-J", "-e") + if err != nil { + return out, err + } + out = ParseClaudeDraft(string(raw)) + out.Agent = agent + return out, nil +} + // Conversation reads the agent's own transcript file (internal/transcript), // and asks the watch to say when it is written to. func (t terminalDriver) Conversation(ctx context.Context, sess Session, all []Session, since int, gen string) (transcript.Result, error) { diff --git a/internal/box/driver_test.go b/internal/box/driver_test.go index b35de23f..70584cd7 100644 --- a/internal/box/driver_test.go +++ b/internal/box/driver_test.go @@ -61,6 +61,15 @@ func (d *recordingDriver) Conversation(_ context.Context, sess Session, _ []Sess return transcript.Result{Source: "driver", Items: []transcript.Item{{Kind: "text", ID: "1", Text: "hello"}}}, nil } +func (d *recordingDriver) Record(_ context.Context, sess Session, _ []Session) (string, string) { + return "", "" +} + +func (d *recordingDriver) Draft(_ context.Context, sess Session) (Draft, error) { + d.note("draft " + sess.Name) + return Draft{Agent: "claude", Text: "writing"}, nil +} + func TestTheAgentIsReachedThroughItsSessionsDriver(t *testing.T) { d := &recordingDriver{} c, _ := servedBox(t, func(b *Box) { b.Drivers = map[string]Driver{DriverTerminal: d} }) diff --git a/internal/box/headless.go b/internal/box/headless.go index 89af9c9b..e7ba1488 100644 --- a/internal/box/headless.go +++ b/internal/box/headless.go @@ -117,7 +117,7 @@ func (h *runHost) headless(ctx context.Context, x *runs.StepCtx) runs.Result { if err := b.beforeAs(ctx, origOf(x), "session.start", map[string]any{"path": dir, "command": command, "agent": agent, "headless": true}); err != nil { return fail(err) } - if _, err := b.Sessions.create(ctx, name, where, dir, command, "", nil, nil); err != nil { + if _, err := b.Sessions.create(ctx, name, where, dir, command, "", nil, nil, nil); err != nil { return fail(err) } b.Events.Publish(eventAs(b, origOf(x), "session.started", map[string]any{"name": name, "location": where, "path": dir, "agent": agent, "headless": true, "run": x.Run.ID})) diff --git a/internal/box/inbox.go b/internal/box/inbox.go index bd208d41..9efc0f2f 100644 --- a/internal/box/inbox.go +++ b/internal/box/inbox.go @@ -62,7 +62,7 @@ func (b *Box) sendQueued(w http.ResponseWriter, r *http.Request) error { if err != nil { return err } - if agent := agentFor(sess); b.atStartupQuestion(ctx, Session{Name: name, Agent: agent}) { + if agent := agentFor(sess); b.atStartupQuestion(ctx, Session{Name: name, Agent: agent, Driver: sess.Driver}) { // Even forced: the question would drop the text and take its Enter. return startupText(agent) } diff --git a/internal/box/inbox_test.go b/internal/box/inbox_test.go index 2de6be8d..252c76b3 100644 --- a/internal/box/inbox_test.go +++ b/internal/box/inbox_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "os" "path/filepath" + "reflect" "strings" "testing" "time" @@ -43,7 +44,7 @@ func TestAPermissionRequestIsKeptOnTheWaitAndNeverPublished(t *testing.T) { want := Ask{Tool: "Bash", Input: "pnpm test --filter checkout", Why: "Run the checkout tests", Message: "Claude needs your permission to use Bash"} st, _ := turns.State("s") - if st.State != "waiting" || st.Ask == nil || *st.Ask != want { + if st.State != "waiting" || st.Ask == nil || !reflect.DeepEqual(*st.Ask, want) { t.Fatalf("state = %+v ask=%+v", st, st.Ask) } for _, e := range seen { diff --git a/internal/box/longprompt_test.go b/internal/box/longprompt_test.go index 0be36902..4500f586 100644 --- a/internal/box/longprompt_test.go +++ b/internal/box/longprompt_test.go @@ -74,7 +74,7 @@ func TestLongPromptsStartAndReadBack(t *testing.T) { command := agent + " " + shellQuote(prompt) name := fmt.Sprintf("task-%d", size>>10) out := filepath.Join(t.TempDir(), "prompt") - sess, err := s.create(ctx, name, "demo/big", t.TempDir(), command, "claude", []string{"OUT=" + out}, nil) + sess, err := s.create(ctx, name, "demo/big", t.TempDir(), command, "claude", []string{"OUT=" + out}, nil, nil) if err != nil { t.Fatalf("create: %v", err) } diff --git a/internal/box/main_test.go b/internal/box/main_test.go index df208500..996456d5 100644 --- a/internal/box/main_test.go +++ b/internal/box/main_test.go @@ -2,15 +2,27 @@ package box import ( "context" + "fmt" "os" "testing" + "github.com/cosscom/shipyard/internal/agenthost" "github.com/cosscom/shipyard/internal/agentpath" ) // Rebases and merges make commits, which git refuses without an identity; // a fresh CI machine has none. func TestMain(m *testing.M) { + // The structured driver's tests run this binary as `berthd agent-host` + // (claudedriver_test.go). + if os.Getenv("BERTH_TEST_AGENT_HOST") == "1" { + code, err := agenthost.Main(os.Args[1:], os.Stdout) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + os.Exit(code) + } for k, v := range map[string]string{ "GIT_AUTHOR_NAME": "berth test", "GIT_AUTHOR_EMAIL": "test@example.com", "GIT_COMMITTER_NAME": "berth test", "GIT_COMMITTER_EMAIL": "test@example.com", diff --git a/internal/box/orchestrate.go b/internal/box/orchestrate.go index 32cf77dc..44e03d74 100644 --- a/internal/box/orchestrate.go +++ b/internal/box/orchestrate.go @@ -199,7 +199,7 @@ func (b *Box) sendPrompt(ctx context.Context, name string, req SendRequest, orig // or Enter, is someone answering. Text meant for the question // (forced, or a key and Enter) is no answer it takes: refused. b.awaitDrawn(ctx, name) - if req.Text != "" && !(isKey(req.Text) && !enter) && b.atStartupQuestion(ctx, Session{Name: name, Agent: agent}) { + if req.Text != "" && !(isKey(req.Text) && !enter) && b.atStartupQuestion(ctx, Session{Name: name, Agent: agent, Driver: sess.Driver}) { if b.Turns == nil || req.Force || isKey(req.Text) { return SendResult{}, startupText(agent) } diff --git a/internal/box/perf_bench_test.go b/internal/box/perf_bench_test.go index bb4632ed..f226ab23 100644 --- a/internal/box/perf_bench_test.go +++ b/internal/box/perf_bench_test.go @@ -76,7 +76,7 @@ func BenchmarkTranscriptPoll(b *testing.B) { b.Setenv("CLAUDE_CONFIG_DIR", cfg) dir := b.TempDir() for i := range 3 { - if _, err := s.create(ctx, fmt.Sprintf("acme-%d", i), "acme", dir, "sleep 600", "claude", nil, nil); err != nil { + if _, err := s.create(ctx, fmt.Sprintf("acme-%d", i), "acme", dir, "sleep 600", "claude", nil, nil, nil); err != nil { b.Fatal(err) } } @@ -116,7 +116,7 @@ func BenchmarkTranscriptPoll(b *testing.B) { func BenchmarkDraftPoll(b *testing.B) { ctx := context.Background() s, execs := benchSessions(b) - if _, err := s.create(ctx, "acme-0", "acme", b.TempDir(), "printf 'acme ready\\n'; sleep 600", "claude", nil, nil); err != nil { + if _, err := s.create(ctx, "acme-0", "acme", b.TempDir(), "printf 'acme ready\\n'; sleep 600", "claude", nil, nil, nil); err != nil { b.Fatal(err) } bx := &Box{Name: "devbox", Sessions: s, Events: &events.Bus{}} @@ -150,7 +150,7 @@ func BenchmarkPollScreens(b *testing.B) { t.init() for i := range 4 { name := fmt.Sprintf("acme-%d", i) - if _, err := s.create(ctx, name, "acme", dir, "sleep 600", "", nil, nil); err != nil { + if _, err := s.create(ctx, name, "acme", dir, "sleep 600", "", nil, nil, nil); err != nil { b.Fatal(err) } t.sess[name] = &sessTrack{Name: name, Agent: "acme-cli", Dir: dir, State: "running", Turns: []*Turn{{ID: name + "#1", Session: name, N: 1, State: "running", Fidelity: "screen"}}} diff --git a/internal/box/secrets.go b/internal/box/secrets.go index a041a3f1..68975d66 100644 --- a/internal/box/secrets.go +++ b/internal/box/secrets.go @@ -623,8 +623,14 @@ func (b *Box) createSession(ctx context.Context, name, location, dir, command st // createAgentSession starts a session for the agent preset agent ("" for // a plain command). func (b *Box) createAgentSession(ctx context.Context, name, location, dir, command, agent string) (Session, error) { + return b.createAgentSessionWith(ctx, name, location, dir, command, agent, nil, nil) +} + +// createAgentSessionWith is createAgentSession with more environment and +// tmux options (a structured agent's: claudedriver.go). +func (b *Box) createAgentSessionWith(ctx context.Context, name, location, dir, command, agent string, extraEnv []string, labels map[string]string) (Session, error) { env, wrap := b.sessionEnv(ctx, dir) - return b.Sessions.create(ctx, name, location, dir, command, agent, env, wrap) + return b.Sessions.create(ctx, name, location, dir, command, agent, append(env, extraEnv...), wrap, labels) } func (b *Box) sessionEnv(ctx context.Context, dir string) (env, wrap []string) { diff --git a/internal/box/sessions.go b/internal/box/sessions.go index 64f8f3ad..d18f8b26 100644 --- a/internal/box/sessions.go +++ b/internal/box/sessions.go @@ -6,10 +6,12 @@ import ( "encoding/base64" "errors" "fmt" + "maps" "os" "os/exec" "path/filepath" "regexp" + "slices" "strconv" "strings" "sync" @@ -73,6 +75,11 @@ type Session struct { // panePID the process its pane runs. commandFile string panePID int + // conv and sock are a structured agent's log folder and its host's + // socket (@berth_conv, @berth_sock; agenthost); bypass, that berthd + // answers its permissions (@berth_bypass). + conv, sock string + bypass bool } var ( @@ -212,7 +219,7 @@ func (s *Sessions) tmux(ctx context.Context, args ...string) ([]byte, error) { // // Sessions started since keep their command in a file (@berth_command_file) // and only its start in @berth_command, for older builds to show. -const listFormat = "#{session_name}\t#{session_created}\t#{session_attached}\t#{@berth_location}\t#{?@berth_command64,,#{@berth_command}}\t#{pane_dead}\t#{pane_start_path}\t#{@berth_command64}\t#{@berth_agent}\t#{@berth_title}\t#{@berth_service}\t#{@berth_command_file}\t#{@berth_scope}\t#{pane_pid}" +const listFormat = "#{session_name}\t#{session_created}\t#{session_attached}\t#{@berth_location}\t#{?@berth_command64,,#{@berth_command}}\t#{pane_dead}\t#{pane_start_path}\t#{@berth_command64}\t#{@berth_agent}\t#{@berth_title}\t#{@berth_service}\t#{@berth_command_file}\t#{@berth_scope}\t#{pane_pid}\t#{@berth_driver}\t#{@berth_conv}\t#{@berth_sock}\t#{@berth_bypass}" // plainCommandMax is how much of a command the plain @berth_command keeps. const plainCommandMax = 1024 @@ -269,10 +276,10 @@ func parseSessions(out []byte) []Session { sessions := []Session{} for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") { f := strings.Split(line, "\t") - for len(f) >= 7 && len(f) < 14 { + for len(f) >= 7 && len(f) < 18 { f = append(f, "") } - if len(f) != 14 { + if len(f) != 18 { continue } command := f[4] @@ -296,9 +303,13 @@ func parseSessions(out []byte) []Session { Title: f[9], Service: f[10], Scope: f[12], + Driver: f[14], commandFile: f[11], panePID: panePID, + conv: f[15], + sock: f[16], + bypass: f[17] == "1", }) } return sessions @@ -308,7 +319,7 @@ func parseSessions(out []byte) []Session { // Commands run through a login shell, so tools the user installed (claude, // codex) are on PATH even when berthd runs under systemd. func (s *Sessions) Create(ctx context.Context, name, location, dir, command string, env []string) (Session, error) { - return s.create(ctx, name, location, dir, command, "", env, nil) + return s.create(ctx, name, location, dir, command, "", env, nil, nil) } // create is Create with the pane's program run behind wrap, a command that @@ -319,7 +330,10 @@ func (s *Sessions) Create(ctx context.Context, name, location, dir, command stri // BERTH_AGENT, so the agent's hooks can say which session they come from, // and agent-browser's idle timeout unless env or berthd's own environment // sets it. -func (s *Sessions) create(ctx context.Context, name, location, dir, command, agent string, env, wrap []string) (Session, error) { +// +// labels are more of the session's tmux options (@berth_driver and the +// like), set with the rest. +func (s *Sessions) create(ctx context.Context, name, location, dir, command, agent string, env, wrap []string, labels map[string]string) (Session, error) { if !sessionName.MatchString(name) { return Session{}, fmt.Errorf("invalid session name %q: use letters, digits, - and _", name) } @@ -388,6 +402,9 @@ func (s *Sessions) create(ctx context.Context, name, location, dir, command, age if scope != "" { set("@berth_scope", scope) } + for _, k := range slices.Sorted(maps.Keys(labels)) { + set(k, labels[k]) + } for i, a := range args { if a != ";" { args[i] = tmuxArg(a) diff --git a/internal/box/sessions_test.go b/internal/box/sessions_test.go index 8999a897..9d819396 100644 --- a/internal/box/sessions_test.go +++ b/internal/box/sessions_test.go @@ -281,7 +281,7 @@ func TestGetAndCaptureIsGetAndAScreen(t *testing.T) { var mu sync.Mutex asked := map[string]int{} s.trace = func(args []string) { mu.Lock(); asked[args[0]]++; mu.Unlock() } - if _, err := s.create(ctx, "both", "acme", t.TempDir(), "printf 'acme ready\\n'; sleep 30", "claude", nil, nil); err != nil { + if _, err := s.create(ctx, "both", "acme", t.TempDir(), "printf 'acme ready\\n'; sleep 30", "claude", nil, nil, nil); err != nil { t.Fatal(err) } want, err := s.Get(ctx, "both") diff --git a/internal/box/startup.go b/internal/box/startup.go index 5be18ca8..1e835f36 100644 --- a/internal/box/startup.go +++ b/internal/box/startup.go @@ -170,7 +170,9 @@ func (b *Box) awaitDrawn(ctx context.Context, name string) int { // question now: from its watch, or, for one no watch follows (berthd // restarted while it asked), from its screen. func (b *Box) atStartupQuestion(ctx context.Context, sess Session) bool { - if sess.Agent == "" { + if sess.Agent == "" || sess.Driver == DriverClaude { + // A structured agent asks nothing at its start: its first prompt + // waits only for its host (startStructured). return false } if v, ok := startups.Load(startKey{b, sess.Name}); ok { diff --git a/internal/box/tasks.go b/internal/box/tasks.go index 8d68a42f..d3e76727 100644 --- a/internal/box/tasks.go +++ b/internal/box/tasks.go @@ -258,6 +258,9 @@ type TaskRequest struct { // names (see AgentPreset); empty is the CLI's default. Model string `json:"model,omitempty"` Effort string `json:"effort,omitempty"` + // Driver and Mode, with an agent: see SessionRequest. + Driver string `json:"driver,omitempty"` + Mode string `json:"mode,omitempty"` // FromSession names the session handing this work off, if any. FromSession string `json:"from_session,omitempty"` // Open asks the app to show the new session: "split" or "tab". @@ -290,14 +293,25 @@ func (b *Box) addTask(w http.ResponseWriter, r *http.Request) error { return err } command := req.Command + var launch *structuredLaunch if command == "" && req.Agent != "" { p, ok := presetFor(&loc, req.Agent) if !ok { return badRequest("unknown agent %q", req.Agent) } - if command, err = AgentCommandWith(p, req.Prompt, req.Model, req.Effort); err != nil { + if structured, err := wantStructured(req.Driver); err != nil { + return err + } else if structured { + l, err := b.launchClaude(p, req.Model, req.Effort, req.Mode, "") + if err != nil { + return err + } + launch, command = &l, p.Command + } else if command, err = AgentCommandWith(p, req.Prompt, req.Model, req.Effort); err != nil { return err } + } else if req.Driver != "" && req.Driver != DriverTerminal { + return badRequest("a driver needs an agent, not a command") } else if req.Model != "" || req.Effort != "" { return badRequest("a model or an effort needs an agent, not a command") } @@ -320,7 +334,12 @@ func (b *Box) addTask(w http.ResponseWriter, r *http.Request) error { if req.Command == "" { preset = req.Agent } - sess, err := b.startSession(r, defaultSessionName(where, command), where, wt.Path, command, preset, preset != "" && req.Prompt != "") + var sess Session + if launch != nil { + sess, err = b.startStructuredSession(r, defaultSessionName(where, command), where, wt.Path, preset, *launch, req.Prompt) + } else { + sess, err = b.startSession(r, defaultSessionName(where, command), where, wt.Path, command, preset, preset != "" && req.Prompt != "") + } if err != nil { // A task is a worktree with an agent in it: without the agent, the // worktree it made goes too, so a retry starts clean. @@ -380,6 +399,11 @@ func (b *Box) enrich(ctx context.Context, all []Session) []Session { continue } s.Driver = b.driverFor(*s).Kind() + if s.Driver == DriverClaude && !s.Exited { + // Followed from the first time it is seen (after a restart, + // the first list), so what it does is said as it does it. + b.claude().conv(*s) + } if s.Exited { if b.Turns != nil { b.Turns.Exited(s.Name) @@ -396,6 +420,16 @@ func (b *Box) enrich(ctx context.Context, all []Session) []Session { s.Queued = st.Queued if s.AgentState == "waiting" { s.Ask = st.Ask + if s.Driver == DriverClaude { + if choices := b.claude().askChoices(*s); len(choices) > 0 { + ask := Ask{Choices: choices} + if s.Ask != nil { + ask = *s.Ask + ask.Choices = choices + } + s.Ask = &ask + } + } } } } diff --git a/internal/box/transcriptapi.go b/internal/box/transcriptapi.go index 6a27d450..2fd2e8b8 100644 --- a/internal/box/transcriptapi.go +++ b/internal/box/transcriptapi.go @@ -90,9 +90,8 @@ func (b *Box) transcript(w http.ResponseWriter, r *http.Request) error { return err } if r.URL.Query().Has("before") { - // Older pages come from the agent's own record, under any driver. - if agent, path, _ := b.transcriptFile(r.Context(), sess, all); path != "" && (agent == "claude" || agent == "codex") { - return olderPage(w, r, agent, path, sess.Dir) + if source, path := b.driverFor(sess).Record(r.Context(), sess, all); path != "" { + return olderPage(w, r, source, path, sess.Dir) } } since, _ := strconv.Atoi(r.URL.Query().Get("since")) diff --git a/internal/box/turns.go b/internal/box/turns.go index 0771fbf6..432f2dda 100644 --- a/internal/box/turns.go +++ b/internal/box/turns.go @@ -74,6 +74,10 @@ type Ask struct { Input string `json:"input,omitempty"` Why string `json:"why,omitempty"` Message string `json:"message,omitempty"` + // Choices are the answers a structured session offers, by the key + // that picks one (claudedriver.go); a terminal's are on its screen. + // Never kept: the session list asks its driver. + Choices []AskChoice `json:"choices,omitempty"` } // SessionState is what an agent session is doing now. @@ -1052,7 +1056,7 @@ func (t *Turns) NoteAsk(data map[string]any, raw any) { return } ask := Ask{Tool: askStr(m, "tool"), Input: askStr(m, "input"), Why: askStr(m, "why"), Message: askStr(m, "message")} - if ask == (Ask{}) { + if ask.Tool == "" && ask.Input == "" && ask.Why == "" && ask.Message == "" { return } t.mu.Lock() diff --git a/internal/box/upgrade.go b/internal/box/upgrade.go index c22d5168..d41ad596 100644 --- a/internal/box/upgrade.go +++ b/internal/box/upgrade.go @@ -162,6 +162,11 @@ func (b *Box) Capabilities() []string { if b.Runs != nil { caps = append(caps, "runs", "exec.detach") } + if b.ConversationsDir != "" { + // driver.claude: sessions and tasks take "driver": "claude", Claude + // Code driven through its stream (claudedriver.go). + caps = append(caps, "driver.claude") + } if b.Artifacts != nil { // artifacts: berthd artifact add, a worktree's artifacts and their // versions (artifactsapi.go), artifact.* events, and the art- diff --git a/internal/integrations/ask_test.go b/internal/integrations/ask_test.go index bc01f435..3eb93734 100644 --- a/internal/integrations/ask_test.go +++ b/internal/integrations/ask_test.go @@ -87,3 +87,19 @@ func TestNotificationMessageIsAnAsk(t *testing.T) { t.Fatalf("ask = %v", e.Data[adapters.AskKey]) } } + +func TestAStructuredAgentsHooksSayNothing(t *testing.T) { + t.Setenv("BERTH_DRIVER", "structured") + var out, errs strings.Builder + emitted := false + Hook([]string{"claude", "Stop", `{"cwd":"/w"}`}, os.Stdin, &out, &errs, func(events.Event) error { + emitted = true + return nil + }) + if emitted || errs.Len() > 0 { + t.Fatalf("the hook spoke for a driven agent: %v %q", emitted, errs.String()) + } + if out.String() != Reply("claude") { + t.Fatalf("the hook's reply = %q", out.String()) + } +} diff --git a/internal/integrations/command.go b/internal/integrations/command.go index cd4521fc..afbc4aa5 100644 --- a/internal/integrations/command.go +++ b/internal/integrations/command.go @@ -40,6 +40,11 @@ func Hook(args []string, stdin *os.File, stdout, stderr io.Writer, emit Emit) { } tool, event := args[0], args[1] defer fmt.Fprint(stdout, Reply(tool)) + if os.Getenv("BERTH_DRIVER") == "structured" { + // berthd drives this agent and reads what it does from its stream: + // its hooks would say it all again. + return + } var payload []byte if len(args) >= 3 { payload = []byte(args[2]) diff --git a/internal/transcript/detail.go b/internal/transcript/detail.go index 91c4c7c6..c3d528f0 100644 --- a/internal/transcript/detail.go +++ b/internal/transcript/detail.go @@ -91,6 +91,10 @@ func Detail(source, path, dir, id string) (ToolDetail, error) { var call, result bool if source == "codex" { call, result = codexDetail(line, id, dir, &d) + } else if source == SourceClaudeHost { + if inner, ok := unwrapHost(line); ok { + call, result = claudeDetail(inner, id, dir, &d) + } } else { call, result = claudeDetail(line, id, dir, &d) } diff --git a/internal/transcript/history.go b/internal/transcript/history.go index 828d8b70..9d7ceb69 100644 --- a/internal/transcript/history.go +++ b/internal/transcript/history.go @@ -28,8 +28,11 @@ const ( ) func parserFor(source string) parser { - if source == "codex" { + switch source { + case "codex": return &codexParser{} + case SourceClaudeHost: + return &hostParser{} } return &claudeParser{} } @@ -74,7 +77,7 @@ func Before(source, path, dir string, before int64, limit int) (Result, error) { items[i].Done, items[i].pending = true, nil } } - return Result{Source: source, Items: append([]Item{}, items...), Crew: []CrewMember{}, More: more}, nil + return Result{Source: shownSource(source), Items: append([]Item{}, items...), Crew: []CrewMember{}, More: more}, nil } } } diff --git a/internal/transcript/host.go b/internal/transcript/host.go new file mode 100644 index 00000000..90034ecb --- /dev/null +++ b/internal/transcript/host.go @@ -0,0 +1,83 @@ +package transcript + +import ( + "bytes" + "encoding/json" + "time" +) + +// SourceClaudeHost is a structured Claude Code session's own log, kept by +// berthd agent-host (internal/agenthost): each line an entry whose data is +// one line of Claude Code's stream-json. Its messages are the lines of +// Claude Code's transcript, so the Claude parser reads them, and the +// conversation reads as the transcript's would. +const SourceClaudeHost = "claude-host" + +// hostEntry is the part of an agent-host log entry read here. +type hostEntry struct { + Dir string `json:"dir"` + At int64 `json:"at"` + Data json.RawMessage `json:"data"` +} + +// hostHead is the part of a stream-json line that says whether it is one +// of the conversation's. +type hostHead struct { + Type string `json:"type"` + Subtype string `json:"subtype"` + Parent *string `json:"parent_tool_use_id"` + Timestamp string `json:"timestamp"` +} + +// unwrapHost is the transcript line a log entry carries: one of the +// conversation's messages, from the agent, with its time. A helper's +// messages (parent_tool_use_id set) are its own, as a subagent's file is. +func unwrapHost(b []byte) ([]byte, bool) { + var e hostEntry + if json.Unmarshal(b, &e) != nil || e.Dir != "out" || len(e.Data) == 0 || e.Data[0] != '{' { + return nil, false + } + var h hostHead + if json.Unmarshal(e.Data, &h) != nil { + return nil, false + } + switch { + case h.Parent != nil && *h.Parent != "": + return nil, false + case h.Type == "user", h.Type == "assistant": + case h.Type == "system" && h.Subtype == "compact_boundary": + default: + return nil, false + } + if h.Timestamp != "" || e.At == 0 { + return e.Data, true + } + // Stream-json lines carry no time of their own: the log's is theirs. + // A key put first is overridden by one of the line's own. + ts, _ := json.Marshal(time.UnixMilli(e.At).UTC().Format(time.RFC3339Nano)) + out := make([]byte, 0, len(e.Data)+len(ts)+16) + out = append(out, `{"timestamp":`...) + out = append(out, ts...) + if rest := bytes.TrimLeft(e.Data[1:], " \t\r\n"); len(rest) > 0 && rest[0] != '}' { + out = append(out, ',') + } + return append(out, e.Data[1:]...), true +} + +// shownSource is the agent a conversation's source is: a structured Claude +// session's log is Claude's conversation. +func shownSource(source string) string { + if source == SourceClaudeHost { + return "claude" + } + return source +} + +// hostParser reads an agent-host log with the Claude parser. +type hostParser struct{ claudeParser } + +func (p hostParser) line(c *conv, b []byte) { + if line, ok := unwrapHost(b); ok { + p.claudeParser.line(c, line) + } +} diff --git a/internal/transcript/host_test.go b/internal/transcript/host_test.go new file mode 100644 index 00000000..d9e9a1e1 --- /dev/null +++ b/internal/transcript/host_test.go @@ -0,0 +1,72 @@ +package transcript + +import ( + "path/filepath" + "strings" + "testing" +) + +// claude_host_ask.jsonl is an agent-host log of a structured Claude Code +// session (recorded, then cut down and made anonymous): a prompt, a +// question the agent asked with its own form and was answered, and its +// reply, among the stream's other lines. +func TestAStructuredSessionsLogReadsAsItsConversation(t *testing.T) { + path := filepath.Join("testdata", "claude_host_ask.jsonl") + res, err := NewReader().Read(SourceClaudeHost, path, "/home/dev/acme", 0) + if err != nil { + t.Fatal(err) + } + if res.Source != "claude" { + t.Fatalf("source %q", res.Source) + } + var kinds []string + for _, it := range res.Items { + kinds = append(kinds, it.Kind) + } + if got := strings.Join(kinds, " "); got != "user question text" { + t.Fatalf("items = %s: %+v", got, res.Items) + } + user, q, reply := res.Items[0], res.Items[1], res.Items[2] + if !strings.HasPrefix(user.Text, "Ask me whether I prefer tea") || user.UUID == "" { + t.Fatalf("the prompt (sent once, echoed once) reads %+v", user) + } + if len(q.Questions) != 1 || q.Questions[0].Question != "Do you prefer Tea or Coffee?" || len(q.Answers) != 1 || q.Answers[0] != "Tea" || !q.Done { + t.Fatalf("question = %+v", q) + } + if reply.Text != "You chose Tea." { + t.Fatalf("reply = %+v", reply) + } + + // A tool call's details and older pages read the same log. + d, err := Detail(SourceClaudeHost, path, "/home/dev/acme", "toolu_01Tct9HfJ6SKZik2AZ5F7FHV") + if err != nil || d.ID == "" { + t.Fatalf("detail = %+v, %v", d, err) + } + page, err := Before(SourceClaudeHost, path, "/home/dev/acme", reply.Off, 10) + if err != nil || len(page.Items) != 2 || page.Items[0].ID != user.ID { + t.Fatalf("older page = %+v, %v", page, err) + } +} + +func TestOnlyTheConversationsOwnLinesAreUnwrapped(t *testing.T) { + for _, tc := range []struct { + line string + ok bool + }{ + {`{"seq":1,"dir":"out","data":{"type":"assistant","message":{"content":"hi"}}}`, true}, + {`{"seq":1,"dir":"out","data":{"type":"assistant","parent_tool_use_id":"toolu_1","message":{"content":"a helper's"}}}`, false}, + {`{"seq":1,"dir":"in","data":{"type":"user","message":{"content":"sent, and echoed when replayed"}}}`, false}, + {`{"seq":1,"dir":"out","data":{"type":"stream_event"}}`, false}, + {`{"seq":1,"dir":"out","data":{"type":"control_request"}}`, false}, + {`{"seq":1,"dir":"out","data":{"type":"system","subtype":"compact_boundary"}}`, true}, + {`{"seq":1,"dir":"out","text":"not json"}`, false}, + } { + if _, ok := unwrapHost([]byte(tc.line)); ok != tc.ok { + t.Errorf("%s: %v", tc.line, ok) + } + } + line, _ := unwrapHost([]byte(`{"seq":1,"at":1791900000000,"dir":"out","data":{"type":"user","message":{"content":"x"}}}`)) + if !strings.HasPrefix(string(line), `{"timestamp":"2026-`) { + t.Fatalf("no time: %s", line) + } +} diff --git a/internal/transcript/testdata/claude_host_ask.jsonl b/internal/transcript/testdata/claude_host_ask.jsonl new file mode 100644 index 00000000..e353c4f8 --- /dev/null +++ b/internal/transcript/testdata/claude_host_ask.jsonl @@ -0,0 +1,11 @@ +{"seq":1,"at":1791900001000,"dir":"host","event":"started","pid":4242,"text":"claude -p --input-format stream-json --output-format stream-json"} +{"seq":2,"at":1791900002000,"dir":"in","data":{"type":"user","message":{"role":"user","content":"Ask me whether I prefer tea or coffee, then tell me what I chose."}}} +{"seq":3,"at":1791900003000,"dir":"out","data":{"type":"system","subtype":"init","cwd":"/home/dev/acme","session_id":"00000000-0000-4000-8000-000000000001","model":"claude-haiku-5-5","permissionMode":"bypassPermissions","slash_commands":["compact","review"],"uuid":"00000000-0000-4000-8000-0000000000aa"}} +{"seq":4,"at":1791900004000,"dir":"out","data":{"type":"user","message":{"role":"user","content":"Ask me whether I prefer tea or coffee, then tell me what I chose."},"session_id":"00000000-0000-4000-8000-000000000001","parent_tool_use_id":null,"uuid":"00000000-0000-4000-8000-0000000000ab","isReplay":true}} +{"seq":5,"at":1791900005000,"dir":"out","data":{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me ask"}},"session_id":"00000000-0000-4000-8000-000000000001","parent_tool_use_id":null}} +{"seq":6,"at":1791900006000,"dir":"out","data":{"type":"assistant","message":{"model":"claude-haiku-5-5","id":"msg_011Cfz4pJP1h3DNYDUBUKz78","type":"message","role":"assistant","content":[{"type":"tool_use","id":"toolu_01Tct9HfJ6SKZik2AZ5F7FHV","name":"AskUserQuestion","input":{"questions":[{"question":"Do you prefer Tea or Coffee?","header":"Drink","options":[{"label":"Tea","description":"You prefer tea."},{"label":"Coffee","description":"You prefer coffee."}],"multiSelect":false}]},"caller":{"type":"direct"}}],"container":null,"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":2,"cache_creation_input_tokens":14476,"cache_read_input_tokens":14660,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":14476},"output_tokens":24,"service_tier":"standard","inference_geo":"global"},"input_transformations":[],"diagnostics":null,"context_management":null},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000001","uuid":"00000000-0000-4000-8000-000000000355","timestamp":"2026-10-13T10:19:20.438Z","request_id":"req_011Cfz4pJ98sy5Z6AkYWjHjk","wire_tool_inputs":{"toolu_01Tct9HfJ6SKZik2AZ5F7FHV":{"questions":[{"question":"Do you prefer Tea or Coffee?","header":"Drink","options":[{"label":"Tea","description":"You prefer tea."},{"label":"Coffee","description":"You prefer coffee."}],"multiSelect":false}]}}}} +{"seq":7,"at":1791900007000,"dir":"out","data":{"type":"control_request","request_id":"00000000-0000-4000-8000-000000000360","request":{"subtype":"can_use_tool","tool_name":"AskUserQuestion","display_name":"AskUserQuestion","input":{"questions":[{"question":"Do you prefer Tea or Coffee?","header":"Drink","options":[{"label":"Tea","description":"You prefer tea."},{"label":"Coffee","description":"You prefer coffee."}],"multiSelect":false}]},"tool_use_id":"toolu_01Tct9HfJ6SKZik2AZ5F7FHV","requires_user_interaction":true}}} +{"seq":8,"at":1791900008000,"dir":"out","data":{"type":"user","message":{"role":"user","content":[{"type":"tool_result","content":"Your questions have been answered: \"Do you prefer Tea or Coffee?\"=\"Tea\". You can now continue with these answers in mind.","tool_use_id":"toolu_01Tct9HfJ6SKZik2AZ5F7FHV"}]},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000001","uuid":"00000000-0000-4000-8000-000000000373","timestamp":"2026-10-13T10:19:20.510Z","tool_use_result":{"questions":[{"question":"Do you prefer Tea or Coffee?","header":"Drink","options":[{"label":"Tea","description":"You prefer tea."},{"label":"Coffee","description":"You prefer coffee."}],"multiSelect":false}],"answers":{"Do you prefer Tea or Coffee?":"Tea"}}}} +{"seq":9,"at":1791900009000,"dir":"out","data":{"type":"assistant","message":{"model":"claude-haiku-5-5","id":"msg_011Cfz4pPa4Uzq8HKffmenKv","type":"message","role":"assistant","content":[{"type":"thinking","thinking":"","signature":""}],"container":null,"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":2,"cache_creation_input_tokens":3063,"cache_read_input_tokens":29136,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":3063},"output_tokens":6,"service_tier":"standard","inference_geo":"global"},"input_transformations":[],"diagnostics":null,"context_management":null},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000001","uuid":"00000000-0000-4000-8000-000000000377","timestamp":"2026-10-13T10:19:21.549Z","request_id":"req_011Cfz4pPHSrct85h5jivWbr","thinking_duration_ms":349,"thinking_display":"updates"}} +{"seq":10,"at":1791900010000,"dir":"out","data":{"type":"assistant","message":{"model":"claude-haiku-5-5","id":"msg_011Cfz4pPa4Uzq8HKffmenKv","type":"message","role":"assistant","content":[{"type":"text","text":"You chose Tea."}],"container":null,"stop_reason":null,"stop_sequence":null,"stop_details":null,"usage":{"input_tokens":2,"cache_creation_input_tokens":3063,"cache_read_input_tokens":29136,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":3063},"output_tokens":6,"service_tier":"standard","inference_geo":"global"},"input_transformations":[],"diagnostics":null,"context_management":null},"parent_tool_use_id":null,"session_id":"00000000-0000-4000-8000-000000000001","uuid":"00000000-0000-4000-8000-000000000378","timestamp":"2026-10-13T10:19:21.585Z","request_id":"req_011Cfz4pPHSrct85h5jivWbr"}} +{"seq":11,"at":1791900011000,"dir":"out","data":{"duration_api_ms":2221,"stop_reason":"end_turn","session_id":"00000000-0000-4000-8000-000000000001","total_cost_usd":0.004052160000000001,"usage":{"input_tokens":4,"cache_creation_input_tokens":17539,"cache_read_input_tokens":43796,"output_tokens":212,"output_tokens_details":{"thinking_tokens":61},"server_tool_use":{"web_search_requests":0,"web_fetch_requests":0},"service_tier":"standard","cache_creation":{"ephemeral_1h_input_tokens":17539,"ephemeral_5m_input_tokens":0},"inference_geo":"global","iterations":[{"input_tokens":2,"output_tokens":70,"cache_read_input_tokens":29136,"cache_creation_input_tokens":3063,"cache_creation":{"ephemeral_5m_input_tokens":0,"ephemeral_1h_input_tokens":3063},"type":"message"}],"speed":"standard","fallback_credit":null},"modelUsage":{"claude-haiku-5-5":{"inputTokens":4,"outputTokens":212,"cacheReadInputTokens":43796,"cacheCreationInputTokens":17539,"webSearchRequests":0,"costUSD":0.004052160000000001,"contextWindow":1000000,"maxOutputTokens":128000,"thinkingTokens":61,"canonicalModel":"claude-haiku-5-5","provider":"firstParty","costBasis":"list"}},"permission_denials":[],"terminal_reason":"completed","fast_mode_state":"off","fast_mode_disabled_reason":"sdk_opt_in_required","subagent_stats":{"spawned":0,"requested":{"background":0,"foreground":0,"unset":0},"started_in_background":0,"max_depth":0,"spawned_by_subagents":0,"completed":0,"failed":0,"killed":{"parent":0,"user":0,"system":0},"refused":{"depth_limit":0,"concurrency_limit":0,"budget":0},"by_type":{}},"safety_stops":0,"is_error":false,"num_turns":2,"subtype":"success","api_error_status":null,"result":"You chose Tea.","ttft_ms":1158,"type":"result","duration_ms":2371,"uuid":"00000000-0000-4000-8000-000000000388","ttft_stream_ms":744,"time_to_request_ms":48,"first_content_frame_ms":745,"queued_turn_count":0,"result_index":0}} diff --git a/internal/transcript/transcript.go b/internal/transcript/transcript.go index 9d74fd2d..0c079220 100644 --- a/internal/transcript/transcript.go +++ b/internal/transcript/transcript.go @@ -470,12 +470,7 @@ func (r *Reader) Follow(source, path, dir string, since int, gen string) (Result r.evictOldest() } c = &conv{source: source, dir: dir, side: isHelper(path), byTool: map[string]int{}, crewByID: map[string]int{}, gen: gens.Add(1)} - switch source { - case "codex": - c.p = &codexParser{} - default: - c.p = &claudeParser{} - } + c.p = parserFor(source) r.convs[path] = c } c.used = now @@ -498,7 +493,7 @@ func (r *Reader) Follow(source, path, dir string, since int, gen string) (Result from = min(from, c.base+n-1) } from = min(from, c.base+len(c.items)) - out := Result{Source: source, Next: c.base + len(c.items), Truncated: c.truncated || c.base > 0, Last: c.lineAt, + out := Result{Source: shownSource(source), Next: c.base + len(c.items), Truncated: c.truncated || c.base > 0, Last: c.lineAt, Gen: c.generation(), Reset: reset, Start: c.offset, File: strings.TrimSuffix(filepath.Base(path), ".jsonl")} if len(c.items) > 0 { out.Start = c.items[0].Off diff --git a/packages/plugin-sdk/src/types.ts b/packages/plugin-sdk/src/types.ts index 3a628167..92b067a3 100644 --- a/packages/plugin-sdk/src/types.ts +++ b/packages/plugin-sdk/src/types.ts @@ -213,8 +213,9 @@ export interface Session { // name. It never runs an agent, and closing its tab never stops it. service?: string; // How the box reaches the agent ("driver" capability): "terminal", its - // own screen in tmux, for every agent today. - driver?: "terminal" | string; + // own screen in tmux, or "claude", Claude Code driven through its stream + // ("driver.claude" capability). + driver?: "terminal" | "claude" | string; // The systemd scope its processes run in, on boxes where sessions get // one (Linux with systemd), and what they use: memory (bytes, page cache // included), its ceiling if the box sets one (Settings → Boxes), processor @@ -233,6 +234,10 @@ export interface Ask { input?: string; why?: string; message?: string; + // The answers a structured session offers, by the key that picks one + // (POST sessions/{name}/send with that key); a terminal's are on its + // screen instead. + choices?: { key: string; label: string }[]; } // A prompt the box holds for an agent until it is idle: the start of its @@ -447,6 +452,12 @@ export interface TaskRequest { // Asks the app to show the new agent: as a tab of its worktree when the // person is looking at it, otherwise as a toast that opens it. open?: "split" | "tab"; + // How berthd runs the agent: "terminal" (the default) or "claude", Claude + // Code structured ("driver.claude" capability); mode is a structured + // agent's permission mode to start in ("bypassPermissions": berthd allows + // what it asks, questions and plans aside). + driver?: "terminal" | "claude"; + mode?: string; } export interface TaskResult {