diff --git a/docs/architecture.md b/docs/architecture.md index 2ec4f439f7..89c2c3ae77 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -593,9 +593,10 @@ port with a generated bearer) or external (`MECATL_BASE_URL`; every local contro surface answers 409 as deployment-owned). Studio is daemon-only — an unreachable daemon renders as an offline state, never demo content — and decodes the wire in one typed seam (`studio/src/lib/protocol/`) that surfaces unknown event kinds -instead of dropping them. Live re-attach to a running session is a stated non-goal -today: the live tail is gRPC-only (`StreamSessionLive`), so Studio shows running -state from the session inventory and reads the transcript when the run ends. A +instead of dropping them. Live re-attach to a running session rides the +durable session watch (`GET /v1/sessions/{id}/watch`, ADR 0250): any tab can +follow a driving run live, with one residual — the driving tab's own +`POST …/prompt` stream still cancels the run on disconnect. A breaking wire change owes a Studio update in the same PR. See ADR 0288/0289. **mecatequi — the single-shot headless runner (`cmd/mecatequi`).** A fourth composition diff --git a/docs/design/PRODUCTION-READINESS.md b/docs/design/PRODUCTION-READINESS.md index 75fdd70116..2df66c662b 100644 --- a/docs/design/PRODUCTION-READINESS.md +++ b/docs/design/PRODUCTION-READINESS.md @@ -46,7 +46,7 @@ record; current behaviour is in the linked [architecture](../architecture.md) do | mecak8s (storage-free k8s-native agent) | ✅ shipped (MVP) · ✅ OPT-IN `/metrics` loopback scrape + OTLP push (ADR 0098) · ✅ verified external Redis TLS/ACL with transactional projected-file reload + last-valid generations (ADR 0240) · ✅ Helm 0.3.0 secure real-provider in-pod TLS+OIDC or edge-terminated TLS+OIDC (ClusterIP h2c), nullable spend ceilings, and pod scheduling controls · ⛔ CRD/Operator · ⛔ HPA (custom-metrics on active-runs) · ⛔ managed Redis provisioning (ElastiCache/MemoryStore — endpoint only) · ⛔ fix `mecated`'s unbounded `GracefulStop` (pre-existing, follow-up) | [mecak8s.md](../adr/0048-mecak8s.md) · [0098](../adr/0098-headless-telemetry.md) · [0240](../adr/0240-mecak8s-credential-reload-and-chart-security.md) · [0278](../adr/0278-mecak8s-edge-terminated-tls.md) · [MECAK8S-PLAN.md](./MECAK8S-PLAN.md) | [overview](../architecture.md) | | ACP adapter (editor stdio surface) | ✅ Phase 1+2 + bounded Phase 3 + multimodal shipped · ⛔ Phase 3 long-tail (rule persistence, grep-over-buffers, fs/* on resume) | [0001-acp-adapter.md](../adr/0001-acp-adapter.md) | [api surface](../architecture/api-surface.md) | | Conversation fork (peer session from a history snapshot) | ✅ shipped · ✅ effort override (mid-conversation effort switch, keeps the transcript — [0068](../adr/0068-effort-change-via-fork.md)) · ⛔ cross-provider/model fork (v2: replay-blob stripping) · ⛔ workspace-branching fork · ⛔ fork-from-event-log-at-arbitrary-point · ⛔ fork lineage (`forked_from` label) | [0065-conversation-fork.md](../adr/0065-conversation-fork.md) | [overview](../architecture.md) | -| Studio (web client) | 🚧 landing as a stacked PR series: ✅ module foundation (vendored Atrium UI kit, toolchain, CI gates) · ✅ server tier (trusted proxy + managed-mode controller core, hermetic suite) · ✅ protocol seam + harness transport · ✅ workspace shell + runtime status · ✅ Chats core + hermetic browser e2e (fixture daemon) · ✅ Scheduled (authoring, carried-spec edit, fire log + per-fire transcripts) · ✅ Skills (browse/create/upload/enable-disable, controller-mediated; learned-skills panel) · ✅ Memory (read-only table + detail + consolidate, honest disabled/empty states) · ✅ Settings core (Personalize, agent identity, learning review) · ✅ provider management + model router + MCP gateway (controller-mediated; server-side auth.yaml ownership, no key material on the wire) · ✅ external-mode OIDC sign-in (PKCE, server-held tokens) · ⛔ advanced chat tiers (attachments, tool drill-down, steer/queue, threads, re-attach, modes, mobile) · ⛔ advanced chat tiers (attachments, steer/queue, threads, re-attach, modes, mobile) | [0288](../adr/0288-studio-atrium-module.md) · [0289](../adr/0289-studio-server-backed-chats.md) | [overview](../architecture.md) | +| Studio (web client) | 🚧 landing as a stacked PR series: ✅ module foundation (vendored Atrium UI kit, toolchain, CI gates) · ✅ server tier (trusted proxy + managed-mode controller core, hermetic suite) · ✅ protocol seam + harness transport · ✅ workspace shell + runtime status · ✅ Chats core + hermetic browser e2e (fixture daemon) · ✅ Scheduled (authoring, carried-spec edit, fire log + per-fire transcripts) · ✅ Skills (browse/create/upload/enable-disable, controller-mediated; learned-skills panel) · ✅ Memory (read-only table + detail + consolidate, honest disabled/empty states) · ✅ Settings core (Personalize, agent identity, learning review) · ✅ provider management + model router + MCP gateway (controller-mediated; server-side auth.yaml ownership, no key material on the wire) · ✅ external-mode OIDC sign-in (PKCE, server-held tokens) · ✅ advanced chat: attachments/previews, tool drill-down + context meter + typed retry + live delegation cards, queue + steer, threads + canvas, live re-attach via the session watch (residual: the driving tab's prompt stream still cancels on disconnect) · ⛔ session modes/model-switch/debug · ⛔ Labs · ⛔ mobile round · ⛔ advanced chat tiers (attachments, steer/queue, threads, re-attach, modes, mobile) | [0288](../adr/0288-studio-atrium-module.md) · [0289](../adr/0289-studio-server-backed-chats.md) | [overview](../architecture.md) | | _Historical / retired_ | — | [ARCHITECTURE.md](../adr/0004-v1-architecture.md) · [STEP-CHAIN.md](../adr/0006-v1-step-chain.md) · [TWELVE-PATTERNS-AUDIT.md](../adr/0007-twelve-patterns-audit.md) · [REPOMAP-TREE-SITTER.md](../adr/0029-repomap-tree-sitter.md) | — | ## Security diff --git a/studio/src/app/workspace/chat/_components/chat-workspace.tsx b/studio/src/app/workspace/chat/_components/chat-workspace.tsx index 594737c094..fe7d491e2a 100644 --- a/studio/src/app/workspace/chat/_components/chat-workspace.tsx +++ b/studio/src/app/workspace/chat/_components/chat-workspace.tsx @@ -395,6 +395,11 @@ export function ChatWorkspace({ sessionId }: { sessionId?: string }) { cancelChat, } = useAgentChat(hookSessionId, { onSessionCreated: handleSessionCreated, + // The inventory poll's lifecycle state: running/awaiting attaches the + // durable watch so an externally-driven run renders live (ADR 0250). + sessionState: hookSessionId + ? sessions.find((s) => s.id === hookSessionId)?.state + : undefined, }); /** Esc with nothing else open interrupts the in-flight run (close.esc). */ diff --git a/studio/src/features/agent/hooks/use-agent-chat.test.ts b/studio/src/features/agent/hooks/use-agent-chat.test.ts index 14cceac30c..61f709e106 100644 --- a/studio/src/features/agent/hooks/use-agent-chat.test.ts +++ b/studio/src/features/agent/hooks/use-agent-chat.test.ts @@ -1,8 +1,9 @@ import { describe, expect, it } from "vitest"; -import type { AgentMessage } from "../types"; +import type { AgentMessage, StreamEvent } from "../types"; import { applyDelegationUpdate, attachmentsFromSteerParts, + reduceWatchEvent, splitPendingSteersOnWatermark, } from "./use-agent-chat"; @@ -36,6 +37,148 @@ describe("splitPendingSteersOnWatermark", () => { }); }); +// ── reduceWatchEvent ───────────────────────────────────────────────────────── + +/** + * Pins the watch transcript reducer (ADR 0250): replayed durable-log events + * rebuild the same message shape the live prompt path produces. + */ +describe("reduceWatchEvent", () => { + let serial = 0; + const nextId = () => `id-${++serial}`; + const run = (events: StreamEvent[]): AgentMessage[] => + events.reduce( + (messages, event) => reduceWatchEvent(messages, event, nextId), + [], + ); + + it("rebuilds a user → assistant exchange with tool activity", () => { + const messages = run([ + { type: "user_prompt", text: "list the files" }, + { type: "token", text: "Sure — " }, + { + type: "tool_call", + callId: "c1", + name: "Bash", + input: "command: ls", + }, + { type: "tool_result", callId: "c1", output: "a.txt", isError: false }, + { type: "token", text: "done." }, + ]); + expect(messages).toHaveLength(2); + expect(messages[0]).toMatchObject({ + role: "user", + content: "list the files", + }); + expect(messages[1]).toMatchObject({ + role: "assistant", + content: "Sure — done.", + toolCalls: [ + { + callId: "c1", + name: "Bash", + output: "a.txt", + status: "completed", + }, + ], + }); + }); + + it("opens a fresh assistant bubble after each user-authored record", () => { + const messages = run([ + { type: "user_prompt", text: "first" }, + { type: "token", text: "answer one" }, + { type: "user_prompt", text: "second" }, + { type: "token", text: "answer two" }, + ]); + expect(messages.map((m) => [m.role, m.content])).toEqual([ + ["user", "first"], + ["assistant", "answer one"], + ["user", "second"], + ["assistant", "answer two"], + ]); + }); + + it("renders a steer echo as a user message, like the committed record it is", () => { + const messages = run([ + { type: "token", text: "working" }, + { type: "steer", text: "focus on tests", messageId: "s-1" }, + { type: "token", text: "ok" }, + ]); + expect(messages.map((m) => [m.role, m.content])).toEqual([ + ["assistant", "working"], + ["user", "focus on tests"], + ["assistant", "ok"], + ]); + }); + + it("renders an approval verdict as a quiet notice line", () => { + const messages = run([ + { + type: "approval_verdict", + approvalId: "a1", + toolName: "Bash", + verdict: "allow_once", + }, + ]); + expect(messages[0].notices).toEqual(["Permission: Bash allowed once"]); + }); + + it("marks a failed terminal on the trailing assistant, failing its running calls", () => { + const messages = run([ + { type: "token", text: "trying" }, + { type: "tool_call", callId: "c9", name: "Edit", input: "" }, + { + type: "run_result", + stop: "error", + text: "", + errorText: "boom", + permanent: true, + }, + ]); + expect(messages[0]).toMatchObject({ + failed: true, + failureDetail: + "boom (permanent — retrying the identical request cannot succeed)", + toolCalls: [{ callId: "c9", status: "failed" }], + }); + }); + + it("fills an empty assistant bubble from a clean terminal's final text", () => { + const messages = run([ + { type: "tool_call", callId: "c2", name: "Read", input: "" }, + { + type: "run_result", + stop: "end_turn", + text: "final", + errorText: "", + permanent: false, + }, + ]); + expect(messages[0].content).toBe("final"); + }); + + it("leaves the transcript untouched for hook-state kinds (asks, usage)", () => { + const before: AgentMessage[] = [ + { id: "m1", role: "assistant", content: "hi", timestamp: 0 }, + ]; + expect( + reduceWatchEvent( + before, + { + type: "approval", + approvalId: "a1", + sessionId: "s", + toolName: "Bash", + description: "", + details: "", + }, + nextId, + ), + ).toBe(before); + }); +}); + // ── delegation cards (D1) ──────────────────────────────────────────────────── describe("applyDelegationUpdate", () => { @@ -136,3 +279,31 @@ describe("attachmentsFromSteerParts", () => { expect(attachmentsFromSteerParts([])).toBeUndefined(); }); }); + +describe("reduceWatchEvent steer parts", () => { + it("keeps the committed steer's media on the rebuilt user bubble", () => { + let serial = 100; + const messages = reduceWatchEvent( + [], + { + type: "steer", + text: "look at this", + messageId: "m-1", + parts: [{ kind: "image", mimeType: "image/png", data: "aGk=" }], + }, + () => `id-${++serial}`, + ); + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ + role: "user", + content: "look at this", + attachments: [ + { + name: "image-1.png", + type: "image/png", + url: "data:image/png;base64,aGk=", + }, + ], + }); + }); +}); diff --git a/studio/src/features/agent/hooks/use-agent-chat.ts b/studio/src/features/agent/hooks/use-agent-chat.ts index bcbef15606..f96a9fab5d 100644 --- a/studio/src/features/agent/hooks/use-agent-chat.ts +++ b/studio/src/features/agent/hooks/use-agent-chat.ts @@ -18,6 +18,7 @@ import { steerHarnessRun, streamHarnessPrompt, } from "@/lib/harness/client"; +import { watchSessionEvents } from "@/lib/harness/watch"; import { encodeSessionPermissionMode, type SessionPermissionMode, @@ -216,6 +217,13 @@ function messagesFromTranscript(transcript: SessionTranscript): AgentMessage[] { return messages; } +/** Quiet human framing for a durable-log approval verdict line. */ +const APPROVAL_VERDICT_LABELS: Record = { + allow_once: "allowed once", + allow_always: "always allowed", + deny: "denied", +}; + /** * Renders the steer drain echo's committed media bundle (ADR 0251) as * message-attachment chips: inline bytes become data: URLs the existing @@ -291,6 +299,187 @@ export function applyDelegationUpdate( return messages; } +/** + * Applies one watch-delivered StreamEvent to a transcript being rebuilt from + * the durable session watch (ADR 0250). Pure and functional (a new array per + * change) so the hook can feed it from replay batches and live frames alike. + * + * The shape mirrors the live prompt path: assistant activity (tokens, tool + * calls, delegations, notices) accumulates onto the TRAILING assistant + * message, and a user-authored record (`user_prompt`, a committed steer + * echo) closes it — the next assistant activity opens a fresh bubble. + * Events with no transcript surface return the list unchanged. + */ +export function reduceWatchEvent( + messages: AgentMessage[], + event: StreamEvent, + nextId: () => string, +): AgentMessage[] { + const last = messages.at(-1); + /** Applies onto the trailing assistant message, opening one if needed. */ + const onAssistant = ( + apply: (message: AgentMessage) => AgentMessage, + ): AgentMessage[] => { + if (last?.role === "assistant") { + return [...messages.slice(0, -1), apply(last)]; + } + return [ + ...messages, + apply({ + id: nextId(), + role: "assistant", + content: "", + timestamp: Date.now(), + }), + ]; + }; + switch (event.type) { + case "user_prompt": { + // The durable record of what the user asked. + if (!event.text) return messages; + return [ + ...messages, + { + id: nextId(), + role: "user", + content: event.text, + timestamp: Date.now(), + }, + ]; + } + case "steer": { + // The committed mid-run steer: text plus its media bundle (ADR 0251), + // so a rebuilt transcript keeps the steer's attachments. + const attachments = attachmentsFromSteerParts(event.parts); + if (!event.text && !attachments) return messages; + return [ + ...messages, + { + id: nextId(), + role: "user", + content: event.text, + timestamp: Date.now(), + attachments, + }, + ]; + } + case "token": + return onAssistant((message) => ({ + ...message, + content: message.content + event.text, + })); + case "reasoning": + return onAssistant((message) => ({ + ...message, + reasoning: (message.reasoning ?? "") + event.text, + })); + case "tool_call": + return onAssistant((message) => ({ + ...message, + toolCalls: [ + ...(message.toolCalls ?? []), + { + callId: event.callId, + name: event.name, + input: event.input, + file: event.file, + status: "running" as const, + }, + ], + })); + case "tool_result": { + // Resolve the matching call on the latest message that carries it. + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (!message.toolCalls?.some((call) => call.callId === event.callId)) { + continue; + } + const updated: AgentMessage = { + ...message, + toolCalls: message.toolCalls.map((call) => + call.callId === event.callId + ? { + ...call, + output: event.output, + isError: event.isError, + status: event.isError + ? ("failed" as const) + : ("completed" as const), + } + : call, + ), + }; + return [ + ...messages.slice(0, index), + updated, + ...messages.slice(index + 1), + ]; + } + return messages; + } + case "approval_verdict": { + // The verdict half of a permission ask, rendered as a quiet one-liner. + const label = APPROVAL_VERDICT_LABELS[event.verdict] ?? event.verdict; + return onAssistant((message) => ({ + ...message, + notices: [ + ...(message.notices ?? []), + `Permission: ${event.toolName || "tool"} ${label || "resolved"}`, + ], + })); + } + case "notice": + return onAssistant((message) => ({ + ...message, + notices: [...(message.notices ?? []), event.text], + })); + case "delegation": + return onAssistant((message) => ({ + ...message, + delegations: [ + ...(message.delegations ?? []), + { + kind: event.kind, + label: event.label, + detail: event.detail, + childId: event.childId, + background: event.background, + routingReason: event.routingReason, + }, + ], + })); + case "delegation_progress": + case "delegation_end": + return applyDelegationUpdate(messages, event); + case "run_result": { + if (event.stop === "error") { + const detail = + event.errorText || "The run failed without a specific error."; + return onAssistant((message) => ({ + ...message, + failed: true, + failureDetail: event.permanent + ? `${detail} (permanent — retrying the identical request cannot succeed)` + : detail, + toolCalls: (message.toolCalls ?? []).map((call) => + call.status === "running" + ? { ...call, status: "failed" as const } + : call, + ), + })); + } + if (event.text && last?.role === "assistant" && !last.content) { + // A run that streamed no deltas still carries its final text here. + return onAssistant((message) => ({ ...message, content: event.text })); + } + return messages; + } + default: + // Approval asks, retractions, and usage are hook state, not transcript. + return messages; + } +} + /** * Chat state for one daemon session. Daemon-only: the sidebar id IS the * daemon session id — there is no client-side session mapping and no demo @@ -309,6 +498,14 @@ export function useAgentChat( /** The composer's pending model pick ("" = auto-routed); read at mint * time like createMode. */ createModel?: () => { modelId: string; providerId: string } | null; + /** + * The session's daemon lifecycle state from the inventory poll + * (idle/running/awaiting/…). Reactive — when it reads running/awaiting + * and the daemon supports `watch_session_events`, the hook attaches a + * durable watch (ADR 0250) to render the externally-driven run live + * instead of a frozen "running" badge. + */ + sessionState?: string; }, ) { const { connected, features, serverCapabilities } = useRuntimeStatus(); @@ -341,6 +538,22 @@ export function useAgentChat( // run_result frames: "retryable" routes retryLast through the retry // endpoint instead of re-sending the prompt. Cleared when a run starts. const lastDispositionRef = useRef(undefined); + // The active run's opaque identity (Event.run_id, ADR 0249), captured from + // the first run-bearing event of the prompt stream — or the latest one a + // watch delivered. Approve/cancel send it as expected_run_id so a stale + // control can never act on the session's NEXT run. "" = unknown. + const runIdRef = useRef(""); + // The session id whose run THIS hook's prompt stream is driving right now — + // the durable watch must not attach on top of it (the prompt path owns the + // view). An id, not a boolean: a stream can outlive a chat switch. + const drivingRef = useRef(null); + // The live watch's teardown + resume position (per-session, opaque). + const watchAbortRef = useRef(null); + const watchCursorRef = useRef(""); + // Set when a watch faulted for this session id: the transcript fallback is + // already showing and re-attaching would loop. Cleared when the session + // leaves the running/awaiting stretch (a later run gets a fresh watch). + const watchFaultedRef = useRef(null); // Attachments sent this visit, keyed by session: the daemon's transcript // carries no attachment bytes, so every rehydrate would strip the chips — // this ref re-attaches them by matching user turns in send order. @@ -357,7 +570,8 @@ export function useAgentChat( /** * Rebuilds the message list from the daemon's authoritative transcript, - * re-attaching this visit's sent-attachment bytes. + * re-attaching this visit's sent-attachment bytes. Shared by the open + * rehydrate, the watch-fault fallback, and the stale-run-control refresh. */ const rehydrate = useCallback(async (id: string, signal?: AbortSignal) => { const transcript = await fetchSessionTranscriptMessages(id, signal); @@ -389,9 +603,21 @@ export function useAgentChat( setMessages(rebuilt); }, []); + // The durable watch is gated on the daemon's open feature registry and the + // session actually having a run to watch (running/awaiting per inventory). + const watchSupported = features.has("watch_session_events"); + const sessionState = options?.sessionState ?? ""; + const watchable = sessionState === "running" || sessionState === "awaiting"; + // Opening a chat (or switching chats) rehydrates from the daemon. + // watchSupported/watchable are deliberately NOT dependencies: they are read + // at open time only — a mid-view flip is the watch effect's business, not a + // reason to refetch (and re-wipe) the transcript. + // biome-ignore lint/correctness/useExhaustiveDependencies: see above useEffect(() => { daemonIdRef.current = sessionId; + runIdRef.current = ""; + watchCursorRef.current = ""; lastDispositionRef.current = undefined; setMessages([]); setPendingApproval(null); @@ -408,6 +634,12 @@ export function useAgentChat( estimatedCost: null, }); if (!sessionId || !connected) return; + // A run another client is driving: the watch effect below owns the + // rebuild (its replay covers the whole transcript), so the fetch here + // would only race it and be overwritten. + if (watchSupported && watchable && drivingRef.current !== sessionId) { + return; + } const controller = new AbortController(); void rehydrate(sessionId, controller.signal).catch((caught) => { if (controller.signal.aborted) return; @@ -417,6 +649,167 @@ export function useAgentChat( return () => controller.abort(); }, [sessionId, connected, rehydrate]); + // The durable session watch (ADR 0250): when this chat's run is being + // driven ELSEWHERE (a schedule fire, another tab, a gRPC client) and the + // daemon supports it, attach from the beginning — the replay rebuilds the + // transcript, the live boundary switches to the streaming view, and the + // terminal result hands back to the normal completed-state flow. Studio's + // own prompt-stream path is untouched: a run this tab drives never watches. + useEffect(() => { + if (!sessionId || !connected || !watchSupported || !watchable) { + // Leaving the running/awaiting stretch clears the fault latch, so the + // session's NEXT run gets a fresh watch. + watchFaultedRef.current = null; + return; + } + if (drivingRef.current === sessionId) return; + if (watchFaultedRef.current === sessionId) return; + const controller = new AbortController(); + watchAbortRef.current = controller; + + // The transcript being rebuilt from replay + live frames. Replay flushes + // in batches (a long history must not commit thousands of renders); every + // live frame renders immediately, like the prompt path. + let rebuilt: AgentMessage[] = []; + let live = false; + let frames = 0; + let serial = 0; + // An ask seen in replay that no later verdict/retract resolved: surfaced + // at the boundary — exactly the parked-approval case (state "awaiting"). + let parkedAsk: ApprovalRequest | null = null; + const nextId = () => { + serial += 1; + return `watch-${serial}`; + }; + const flush = () => setMessages(rebuilt); + const resolveAsk = (approvalId: string) => { + if (parkedAsk?.approvalId === approvalId) parkedAsk = null; + if (!live) return; + setPendingApproval((current) => + current?.approvalId === approvalId ? null : current, + ); + setStatus((current) => + current === "waiting_approval" ? "streaming" : current, + ); + }; + + setStatus("streaming"); + setPendingApproval(null); + setError(null); + + void watchSessionEvents( + sessionId, + (delivery) => { + if (delivery.cursor) watchCursorRef.current = delivery.cursor; + const event = delivery.event; + if (!event) { + if (delivery.phase === "live" && !live) { + // The replay→live boundary: the rebuild is complete. Show it and + // surface a still-unresolved ask (the parked-approval case). + live = true; + flush(); + if (parkedAsk) { + setPendingApproval(parkedAsk); + setStatus("waiting_approval"); + } + } + return; + } + // The LATEST run-bearing event names the current run (a replay spans + // every earlier run of the session too). + if (event.runId) runIdRef.current = event.runId; + switch (event.type) { + case "approval": + parkedAsk = { + approvalId: event.approvalId, + sessionId, + toolName: event.toolName, + description: event.description, + details: event.details, + }; + if (live) { + setPendingApproval(parkedAsk); + setStatus("waiting_approval"); + } + break; + case "retract": + resolveAsk(event.approvalId); + break; + case "approval_verdict": + // Another client resolved the ask; the quiet verdict line also + // lands in the transcript via the reducer. + resolveAsk(event.approvalId); + rebuilt = reduceWatchEvent(rebuilt, event, nextId); + if (live) flush(); + break; + case "usage": + // Only live frames accumulate: replay covers finished runs whose + // figures this visit never counted anywhere else either. + if (live) { + setUsage((prev) => ({ + inputTokens: prev.inputTokens + event.inputTokens, + outputTokens: prev.outputTokens + event.outputTokens, + cacheReadTokens: + prev.cacheReadTokens + (event.cacheReadTokens ?? 0), + cacheWriteTokens: + prev.cacheWriteTokens + (event.cacheWriteTokens ?? 0), + reasoningTokens: + prev.reasoningTokens + (event.reasoningTokens ?? 0), + estimatedCost: event.estimatedCost, + })); + } + break; + case "run_result": + rebuilt = reduceWatchEvent(rebuilt, event, nextId); + if (live) { + // The terminal result ends the watch; the normal + // completed-state flow takes over from here. + flush(); + runIdRef.current = ""; + if (event.stop === "error") { + lastDispositionRef.current = event.retryDisposition; + setError( + event.errorText || "The run failed without a specific error.", + ); + setStatus("error"); + } else { + setStatus("idle"); + } + controller.abort(); + } + break; + default: + rebuilt = reduceWatchEvent(rebuilt, event, nextId); + frames += 1; + if (live || frames % 200 === 0) flush(); + } + }, + { signal: controller.signal }, + ).catch(() => { + if (controller.signal.aborted) return; + // Any watch fault — activity_gap / cursor_expired / an exhausted + // reconnect budget / a pre-stream refusal — falls back to the + // authoritative transcript: the codes differ, the recovery is the + // same, and the latch stops a re-attach loop while the run continues. + watchFaultedRef.current = sessionId; + setPendingApproval(null); + setStatus("idle"); + void rehydrate(sessionId).catch(() => undefined); + }); + + return () => { + controller.abort(); + if (watchAbortRef.current === controller) watchAbortRef.current = null; + // A torn-down watch (chat switch, state flip, disconnect) must not + // leave the streaming badge stuck; real terminals set their own state. + setStatus((current) => + current === "streaming" || current === "waiting_approval" + ? "idle" + : current, + ); + }; + }, [sessionId, connected, watchSupported, watchable, rehydrate]); + const queueMessage = useCallback((text: string, files?: File[]) => { const trimmed = text.trim(); if (!trimmed && !files?.length) return; @@ -448,6 +841,11 @@ export function useAgentChat( ), ); return (event: StreamEvent) => { + // The first run-bearing event names the run (ADR 0249); the id + // scopes this run's approve/cancel/steer controls. + if (event.runId && !runIdRef.current) { + runIdRef.current = event.runId; + } switch (event.type) { case "token": patch((message) => ({ @@ -610,6 +1008,8 @@ export function useAgentChat( })); break; case "run_result": + // The run is over; a control scoped to it would be stale. + runIdRef.current = ""; if (event.stop === "error") { // The typed disposition routes the Retry button (ADR 0239). lastDispositionRef.current = event.retryDisposition; @@ -761,6 +1161,10 @@ export function useAgentChat( } onSessionCreatedRef.current?.(daemonId); } + // This tab drives the run now: the durable watch must not attach on + // top of the prompt stream, and the run's identity starts unknown. + drivingRef.current = daemonId; + runIdRef.current = ""; lastDispositionRef.current = undefined; await streamHarnessPrompt( daemonId, @@ -798,6 +1202,9 @@ export function useAgentChat( })); } finally { abortRef.current = null; + if (daemonId && drivingRef.current === daemonId) { + drivingRef.current = null; + } } }, [status, connected, queueMessage, makeStreamHandler], @@ -887,6 +1294,9 @@ export function useAgentChat( ]); const controller = new AbortController(); abortRef.current = controller; + // This tab drives the retried run: the watch must not attach on top. + drivingRef.current = daemonId; + runIdRef.current = ""; lastDispositionRef.current = undefined; let ineligible = false; try { @@ -921,6 +1331,7 @@ export function useAgentChat( } } finally { abortRef.current = null; + if (drivingRef.current === daemonId) drivingRef.current = null; } if (ineligible) await resendLast(); }, [status, connected, resendLast, makeStreamHandler]); @@ -948,7 +1359,13 @@ export function useAgentChat( const cancelChat = useCallback(async () => { abortRef.current?.abort(); if (daemonIdRef.current) { - await cancelHarnessRun(daemonIdRef.current); + // Scoped to the run this hook knows about (ADR 0249): if that run + // already ended, the daemon answers 409 stale_run_control and the + // session's NEXT run is left untouched — exactly what "cancel" meant. + await cancelHarnessRun( + daemonIdRef.current, + runIdRef.current || undefined, + ); } setStatus("idle"); }, []); @@ -962,16 +1379,22 @@ export function useAgentChat( /** * Injects a message — text plus staged image attachments (ADR 0251) — into * the in-flight run at the next turn boundary. accepted/appended park it on - * the pending list until the drain echo; too_late or a failed request fall - * back to the queue so nothing is lost — the queue drains it as a normal - * prompt. + * the pending list until the drain echo; too_late, a 409 + * `stale_run_control` (the strict steer's "that run already ended"), or a + * failed request fall back to the queue so nothing is lost — the queue + * drains it as a normal prompt. + * + * Every steer is STRICT (expected_run_id, ADR 0252): an unqualified steer + * that loses the terminal race would be PROMOTED into a follow-up run + * behind Studio's back, so an unknown run id queues rather than steers. */ const steerMessage = useCallback( async (text: string, files?: File[]) => { const trimmed = text.trim(); if (!trimmed && !files?.length) return; const daemonId = daemonIdRef.current; - if (!daemonId || !steerSupported) { + const runId = runIdRef.current; + if (!daemonId || !steerSupported || !runId) { queueMessage(trimmed, files); return; } @@ -1004,6 +1427,7 @@ export function useAgentChat( const id = `steer-${Date.now()}-${steerSerialRef.current}`; try { const { outcome } = await steerHarnessRun(daemonId, trimmed, id, { + expectedRunId: runId, parts, }); if (outcome === "accepted" || outcome === "appended") { @@ -1028,6 +1452,16 @@ export function useAgentChat( } queueMessage(trimmed, files); } catch (caught) { + if ( + caught instanceof HarnessApiError && + caught.code === "stale_run_control" + ) { + // The named run already ended — exactly the too_late outcome: + // requeue quietly, never an error toast (A2.3/C1.3). + runIdRef.current = ""; + queueMessage(trimmed, files); + return; + } queueMessage(trimmed, files); setError(caught instanceof Error ? caught.message : String(caught)); } @@ -1146,13 +1580,26 @@ export function useAgentChat( : choice === "once" ? "allow_once" : "allow_always", + runIdRef.current || undefined, ); } catch (caught) { + if ( + caught instanceof HarnessApiError && + caught.code === "stale_run_control" + ) { + // The run this dialog belonged to already ended (e.g. another + // client answered, or a schedule fire replaced it). Not an error: + // refresh quietly and let the transcript show what happened. + runIdRef.current = ""; + setStatus("idle"); + void rehydrate(daemonId).catch(() => undefined); + return; + } setError(caught instanceof Error ? caught.message : String(caught)); setStatus("error"); } }, - [pendingApproval], + [pendingApproval, rehydrate], ); const respondToClarification = useCallback(async (_response: string) => { diff --git a/user-docs/building/what-you-get/studio.md b/user-docs/building/what-you-get/studio.md index 4413a4eeae..f694069caf 100644 --- a/user-docs/building/what-you-get/studio.md +++ b/user-docs/building/what-you-get/studio.md @@ -12,8 +12,8 @@ Studio reads and writes the daemon's state rather than keeping its own. :::note Landing in progress Studio is landing as a stacked series of pull requests. This page grows with -each one; right now everything except the advanced chat tiers (attachments, steering, -threads, live re-attach, modes, mobile) is in the tree. +each one; right now everything except session modes/model-switch/debug, Labs, and the +mobile round is in the tree. ::: ## Starting it @@ -110,3 +110,10 @@ provider page offers sign-in: Studio's server tier runs the PKCE flow itself and holds tokens in process memory only — nothing token-shaped reaches the browser or disk. With OIDC unconfigured, the static `MECATL_AUTH_TOKEN` path is untouched. Managed mode never uses either. + +## Live re-attach + +Open a driving session in a second tab (or refresh mid-run) and Studio +re-attaches to the live run through the daemon's durable session watch. One +limit to know: the tab that *started* the run still cancels it if that tab's +prompt connection drops — re-attach covers every other tab.