From 8f0ae08617347da01f68f558df41df6156af1968 Mon Sep 17 00:00:00 2001 From: jtenniswood Date: Wed, 2 Sep 2026 11:25:38 +0100 Subject: [PATCH] =?UTF-8?q?feat(studio):=20harness=20transport=20=E2=80=94?= =?UTF-8?q?=20daemon=20+=20controller=20client=20modules?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit src/lib/harness lands whole and final: client.ts (the transport monolith — daemon plane, controller plane, and the RFC-9457 HarnessApiError machinery every satellite assumes), wire.ts, watch.ts (the ADR 0250 SSE grammar with cursor-resumed reconnects), and the learning / learned-skills / dream / storage / debug satellites, each with its colocated vitest suite (server-info.ts waits for its only consumer, the About card, later in the series). One deliberate addition over the original branch: client-surface.test.ts, an export-pin smoke test enumerating the transport surface. Controller-plane functions whose UI consumers land later in the series exist here without a caller; the pin makes that surface knip-visible through a test entry — a permanent, honest guard against export drift instead of a temporary ignore. Co-Authored-By: Claude Fable 5 --- studio/src/lib/harness/client-actions.test.ts | 240 +++ studio/src/lib/harness/client-errors.test.ts | 94 + studio/src/lib/harness/client-surface.test.ts | 142 ++ studio/src/lib/harness/client.ts | 1545 +++++++++++++++++ studio/src/lib/harness/create-body.test.ts | 129 ++ studio/src/lib/harness/debug.test.ts | 73 + studio/src/lib/harness/debug.ts | 56 + studio/src/lib/harness/dream.test.ts | 161 ++ studio/src/lib/harness/dream.ts | 178 ++ studio/src/lib/harness/learned-skills.test.ts | 185 ++ studio/src/lib/harness/learned-skills.ts | 244 +++ studio/src/lib/harness/learning.test.ts | 190 ++ studio/src/lib/harness/learning.ts | 217 +++ studio/src/lib/harness/storage.test.ts | 89 + studio/src/lib/harness/storage.ts | 62 + studio/src/lib/harness/watch.test.ts | 196 +++ studio/src/lib/harness/watch.ts | 276 +++ studio/src/lib/harness/wire.ts | 50 + studio/src/lib/protocol/index.ts | 24 + studio/src/lib/protocol/sessions.ts | 2 +- 20 files changed, 4152 insertions(+), 1 deletion(-) create mode 100644 studio/src/lib/harness/client-actions.test.ts create mode 100644 studio/src/lib/harness/client-errors.test.ts create mode 100644 studio/src/lib/harness/client-surface.test.ts create mode 100644 studio/src/lib/harness/client.ts create mode 100644 studio/src/lib/harness/create-body.test.ts create mode 100644 studio/src/lib/harness/debug.test.ts create mode 100644 studio/src/lib/harness/debug.ts create mode 100644 studio/src/lib/harness/dream.test.ts create mode 100644 studio/src/lib/harness/dream.ts create mode 100644 studio/src/lib/harness/learned-skills.test.ts create mode 100644 studio/src/lib/harness/learned-skills.ts create mode 100644 studio/src/lib/harness/learning.test.ts create mode 100644 studio/src/lib/harness/learning.ts create mode 100644 studio/src/lib/harness/storage.test.ts create mode 100644 studio/src/lib/harness/storage.ts create mode 100644 studio/src/lib/harness/watch.test.ts create mode 100644 studio/src/lib/harness/watch.ts create mode 100644 studio/src/lib/harness/wire.ts create mode 100644 studio/src/lib/protocol/index.ts diff --git a/studio/src/lib/harness/client-actions.test.ts b/studio/src/lib/harness/client-actions.test.ts new file mode 100644 index 0000000000..6caa633452 --- /dev/null +++ b/studio/src/lib/harness/client-actions.test.ts @@ -0,0 +1,240 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { StreamEvent } from "@/features/agent/types"; +import { + cancelHarnessSteer, + compactHarnessSession, + fetchHarnessSessionDetail, + listHarnessAgents, + retryHarnessRun, + steerHarnessRun, +} from "./client"; + +/** + * Pins the request/response contracts of the chat-resilience client calls: + * manual compaction (ADR 0244), the strict multimodal steer + the + * cancel-steer route naming (ADR 0252), and the GET-session resolved-model + * echo the context meter reads (B1). + */ + +type Captured = { url: string; init?: RequestInit }; + +function stubFetch(status: number, body: unknown): Captured { + const captured: Captured = { url: "" }; + vi.stubGlobal("fetch", async (url: RequestInfo | URL, init?: RequestInit) => { + captured.url = String(url); + captured.init = init; + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); + }); + return captured; +} + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("compactHarnessSession", () => { + it("POSTs bodyless and reads the compacted bool", async () => { + const captured = stubFetch(200, { compacted: true }); + await expect(compactHarnessSession("s1")).resolves.toBe(true); + expect(captured.url).toBe("/api/mecatl/v1/sessions/s1/compact"); + expect(captured.init?.method).toBe("POST"); + expect(captured.init?.body).toBeUndefined(); + }); + + it("reads an empty answer as nothing-to-compact (the live daemon omits false)", async () => { + stubFetch(200, {}); + await expect(compactHarnessSession("s1")).resolves.toBe(false); + }); + + it("throws the typed error on a refusal", async () => { + stubFetch(412, { code: "session_active", error: "session is running" }); + await expect(compactHarnessSession("s1")).rejects.toMatchObject({ + name: "HarnessApiError", + status: 412, + code: "session_active", + }); + }); +}); + +describe("steerHarnessRun", () => { + it("sends the strict body: text, message_id, parts, and expected_run_id (ADR 0252)", async () => { + const captured = stubFetch(200, { + outcome: "accepted", + message_id: "m-1", + }); + const result = await steerHarnessRun("s1", "focus", "m-1", { + expectedRunId: "run-9", + parts: [{ kind: "image", mime_type: "image/png", data: "aGk=" }], + }); + expect(result).toEqual({ outcome: "accepted", messageId: "m-1" }); + expect(captured.url).toBe("/api/mecatl/v1/sessions/s1/steer"); + expect(JSON.parse(String(captured.init?.body))).toEqual({ + text: "focus", + message_id: "m-1", + expected_run_id: "run-9", + parts: [{ kind: "image", mime_type: "image/png", data: "aGk=" }], + }); + }); + + it("omits parts and expected_run_id when the caller has none (legacy shape)", async () => { + const captured = stubFetch(200, { outcome: "appended" }); + await steerHarnessRun("s1", "focus", "m-2"); + expect(JSON.parse(String(captured.init?.body))).toEqual({ + text: "focus", + message_id: "m-2", + }); + }); + + it("surfaces the strict 409 as the typed stale_run_control error", async () => { + stubFetch(409, { + code: "stale_run_control", + error: "the named run already ended", + }); + await expect( + steerHarnessRun("s1", "focus", "m-3", { expectedRunId: "run-old" }), + ).rejects.toMatchObject({ + name: "HarnessApiError", + status: 409, + code: "stale_run_control", + }); + }); +}); + +describe("cancelHarnessSteer", () => { + it("POSTs the ADR-0252 cancel-steer route (steer-cancel is the deprecated alias)", async () => { + const captured = stubFetch(200, { outcome: "retracted" }); + await expect(cancelHarnessSteer("s1")).resolves.toBe("retracted"); + expect(captured.url).toBe("/api/mecatl/v1/sessions/s1/cancel-steer"); + expect(captured.init?.method).toBe("POST"); + }); +}); + +describe("retryHarnessRun", () => { + it("POSTs the retry route bodyless and relays the SSE exactly like a prompt (B2.2)", async () => { + const sse = [ + 'data: {"type":"model.retry","model_retry":{"retry_disposition":2}}', + "", + 'data: {"type":"message.delta","text":"resumed"}', + "", + 'data: {"type":"result","result":{"stop":"end_turn","text":"done"}}', + "", + "", + ].join("\n"); + const captured: Captured = { url: "" }; + vi.stubGlobal( + "fetch", + async (url: RequestInfo | URL, init?: RequestInit) => { + captured.url = String(url); + captured.init = init; + return new Response(sse, { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); + }, + ); + const events: StreamEvent[] = []; + await retryHarnessRun("s1", (event) => events.push(event)); + expect(captured.url).toBe("/api/mecatl/v1/sessions/s1/retry"); + expect(captured.init?.method).toBe("POST"); + expect(captured.init?.body).toBeUndefined(); + expect(events).toEqual([ + { type: "notice", text: "Retrying the failed step…" }, + { type: "token", text: "resumed" }, + { + type: "run_result", + stop: "end_turn", + text: "done", + errorText: "", + permanent: false, + }, + ]); + }); + + it("surfaces the 409 ineligibility as the typed code", async () => { + stubFetch(409, { + code: "failed_step_retry_ineligible", + error: "retry is not eligible", + }); + await expect(retryHarnessRun("s1", () => {})).rejects.toMatchObject({ + name: "HarnessApiError", + status: 409, + code: "failed_step_retry_ineligible", + }); + }); +}); + +describe("fetchHarnessSessionDetail", () => { + it("decodes the resolved_model echo and the capabilities object (B1)", async () => { + stubFetch(200, { + session_id: "s1", + resolved_model: { + provider_id: "openrouter", + model_id: "openai/gpt-5", + context_window: 400000, + }, + capabilities: { manual_compaction: true }, + }); + await expect(fetchHarnessSessionDetail("s1")).resolves.toEqual({ + resolvedModel: { + providerId: "openrouter", + modelId: "openai/gpt-5", + contextWindow: 400000, + }, + capabilities: { manual_compaction: true }, + }); + }); + + it("tolerates a daemon that echoes neither field", async () => { + stubFetch(200, { session_id: "s1", mode: "default" }); + await expect(fetchHarnessSessionDetail("s1")).resolves.toEqual({ + resolvedModel: null, + capabilities: {}, + }); + }); +}); + +describe("listHarnessAgents", () => { + it("decodes the full AgentInfo row: model, tools, permission_mode, color (D2.2)", async () => { + stubFetch(200, { + agents: [ + { + name: "reviewer", + description: "Reviews diffs", + model: "anthropic/claude-opus-4.6", + tools: ["Read", "Grep", 7, "Glob"], + permission_mode: "plan", + color: "cyan", + }, + // A minimal def: the live daemon omits every empty field. + { name: "explorer" }, + ], + }); + await expect(listHarnessAgents()).resolves.toEqual([ + { + name: "reviewer", + description: "Reviews diffs", + model: "anthropic/claude-opus-4.6", + // Non-string entries are dropped, never rendered. + tools: ["Read", "Grep", "Glob"], + permissionMode: "plan", + color: "cyan", + }, + { + name: "explorer", + description: "", + model: "", + tools: [], + permissionMode: "", + color: "", + }, + ]); + }); + + it("reads an empty roster off the live daemon's bare {} response", async () => { + stubFetch(200, {}); + await expect(listHarnessAgents()).resolves.toEqual([]); + }); +}); diff --git a/studio/src/lib/harness/client-errors.test.ts b/studio/src/lib/harness/client-errors.test.ts new file mode 100644 index 0000000000..228e7b3ffa --- /dev/null +++ b/studio/src/lib/harness/client-errors.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, it } from "vitest"; +import { fetchHarnessCompatibility, HarnessApiError } from "./client"; + +/** + * Pins the ADR-0248 client contract: errors are typed on the stable machine + * `code` (problem-details body), with the legacy `error` prose as the + * message; named codes get plainer framing; compatibility decoding tolerates + * older daemons (404 → null). + */ + +function jsonResponse(status: number, body: unknown): Response { + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/problem+json" }, + }); +} + +describe("HarnessApiError via fetchHarnessCompatibility", () => { + it("carries the stable code and the server's message", async () => { + const original = globalThis.fetch; + globalThis.fetch = async () => + jsonResponse(409, { + type: "urn:mecatl:error:stale_run_control", + code: "stale_run_control", + error: "the named run already ended", + status: 409, + }); + try { + await expect(fetchHarnessCompatibility()).rejects.toMatchObject({ + name: "HarnessApiError", + status: 409, + code: "stale_run_control", + message: "the named run already ended", + }); + } finally { + globalThis.fetch = original; + } + }); + + it("frames the named codes in plain language", async () => { + const original = globalThis.fetch; + globalThis.fetch = async () => + jsonResponse(503, { code: "draining", error: "server draining" }); + try { + const error = await fetchHarnessCompatibility().catch((e) => e); + expect(error).toBeInstanceOf(HarnessApiError); + expect((error as HarnessApiError).code).toBe("draining"); + expect((error as HarnessApiError).message).toMatch(/restarting/); + } finally { + globalThis.fetch = original; + } + }); + + it("degrades to status text against a non-JSON error body", async () => { + const original = globalThis.fetch; + globalThis.fetch = async () => + new Response("nope", { status: 500, statusText: "Internal Error" }); + try { + const error = await fetchHarnessCompatibility().catch((e) => e); + expect((error as HarnessApiError).code).toBe(""); + expect((error as HarnessApiError).message).toBe("500 Internal Error"); + } finally { + globalThis.fetch = original; + } + }); + + it("decodes the compatibility document and tolerates a 404 daemon", async () => { + const original = globalThis.fetch; + globalThis.fetch = async () => + jsonResponse(200, { + api_major: 1, + features: ["watch_session_events", 7, "server_info"], + capabilities: { steer: true }, + deployment: "lab-1", + }); + try { + const doc = await fetchHarnessCompatibility(); + expect(doc).toEqual({ + apiMajor: 1, + features: ["watch_session_events", "server_info"], + capabilities: { steer: true }, + deployment: "lab-1", + }); + } finally { + globalThis.fetch = original; + } + globalThis.fetch = async () => new Response("", { status: 404 }); + try { + expect(await fetchHarnessCompatibility()).toBeNull(); + } finally { + globalThis.fetch = original; + } + }); +}); diff --git a/studio/src/lib/harness/client-surface.test.ts b/studio/src/lib/harness/client-surface.test.ts new file mode 100644 index 0000000000..c6b0aa65c0 --- /dev/null +++ b/studio/src/lib/harness/client-surface.test.ts @@ -0,0 +1,142 @@ +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it } from "vitest"; +import type { HarnessResolvedModel, HarnessRouterCategory } from "./client"; +import { + apiError, + cancelHarnessRun, + cancelHarnessSteer, + compactHarnessSession, + connectHarnessGateway, + createHarnessSession, + createHarnessSkill, + createHarnessSkillFiles, + createThreadHarnessSession, + deleteHarnessSession, + deleteHarnessSkill, + fetchAllSessions, + fetchHarnessCompatibility, + fetchHarnessControlStatus, + fetchHarnessRouter, + fetchHarnessSessionDetail, + fetchHarnessSessionMode, + fetchHarnessSkillBody, + fetchHarnessSkillFile, + fetchHarnessUserModel, + fetchSessionTranscriptMessages, + forkHarnessSessionToModel, + HARNESS_API, + HarnessApiError, + harnessScheduleAction, + listDisabledHarnessSkills, + listHarnessAgents, + listHarnessCommands, + listHarnessModels, + listHarnessProviders, + listHarnessSkillFiles, + listHarnessSkills, + listKnownHarnessProviders, + listScheduleFires, + listScheduleRows, + probeHarness, + removeHarnessProvider, + renameHarnessSession, + respondToHarnessApproval, + restartHarnessDaemon, + retryHarnessRun, + saveHarnessRouter, + saveHarnessSchedule, + saveHarnessSkillBody, + setActiveHarnessProvider, + setHarnessSessionMode, + setHarnessSkillEnabled, + startHarnessGatewayOAuth, + steerHarnessRun, + streamHarnessPrompt, + ThreadSourceBusyError, + testHarnessProviderKey, + waitForHarnessGateway, +} from "./client"; +import { fetchLearnedSkill } from "./learned-skills"; +import { undoLearningPromotion } from "./learning"; + +/** + * Export-surface pin for the transport monolith. client.ts lands whole and + * final ahead of some of its UI consumers (they arrive later in the stacked + * series that split PR #618), so knip cannot see every export consumed yet. + * This test is the honest, PERMANENT guard that replaces a temporary knip + * ignore: every public value export is enumerated here, so an accidental + * export rename/removal fails vitest, and knip counts each as consumed by a + * test entry. Removing a genuine export means updating this list — a visible + * decision, exactly like the harness-token exemption maps on the Go side. + */ +describe("client.ts public surface", () => { + it("exports every transport entry point", () => { + // Type-only surface consumed by later PRs in the series (model router, + // resolved-model echo) — referencing them here keeps knip honest. + const typeSurface: { + category?: HarnessRouterCategory; + resolved?: HarnessResolvedModel; + } = {}; + expect(typeSurface).toBeDefined(); + const surface = { + fetchLearnedSkill, + undoLearningPromotion, + HARNESS_API, + HarnessApiError, + ThreadSourceBusyError, + apiError, + cancelHarnessRun, + cancelHarnessSteer, + compactHarnessSession, + connectHarnessGateway, + createHarnessSession, + createHarnessSkill, + createHarnessSkillFiles, + createThreadHarnessSession, + deleteHarnessSession, + deleteHarnessSkill, + fetchAllSessions, + fetchHarnessCompatibility, + fetchHarnessControlStatus, + fetchHarnessRouter, + fetchHarnessSessionDetail, + fetchHarnessSessionMode, + fetchHarnessSkillBody, + fetchHarnessSkillFile, + fetchHarnessUserModel, + fetchSessionTranscriptMessages, + forkHarnessSessionToModel, + harnessScheduleAction, + listDisabledHarnessSkills, + listHarnessAgents, + listHarnessCommands, + listHarnessModels, + listHarnessProviders, + listHarnessSkillFiles, + listHarnessSkills, + listKnownHarnessProviders, + listScheduleFires, + listScheduleRows, + probeHarness, + removeHarnessProvider, + renameHarnessSession, + respondToHarnessApproval, + restartHarnessDaemon, + retryHarnessRun, + saveHarnessRouter, + saveHarnessSchedule, + saveHarnessSkillBody, + setActiveHarnessProvider, + setHarnessSessionMode, + setHarnessSkillEnabled, + startHarnessGatewayOAuth, + steerHarnessRun, + streamHarnessPrompt, + testHarnessProviderKey, + waitForHarnessGateway, + }; + for (const [name, value] of Object.entries(surface)) { + expect(value, name).toBeDefined(); + } + }); +}); diff --git a/studio/src/lib/harness/client.ts b/studio/src/lib/harness/client.ts new file mode 100644 index 0000000000..87821221ce --- /dev/null +++ b/studio/src/lib/harness/client.ts @@ -0,0 +1,1545 @@ +/** + * Browser-side client for the mecatl daemon, reached through the same-origin + * /api/mecatl proxy (which injects auth and the workspace server-side). + * + * All wire decoding lives in the protocol seam (src/lib/protocol); this + * module owns transport: fetch calls, SSE frame buffering, and the stream + * robustness rules (idle timeout, terminal-result guard). + */ + +import type { StreamEvent } from "@/features/agent/types"; +import { validSkillName } from "@/lib/controller-security.mjs"; +import { + decodeScheduleFires, + decodeScheduleRows, + decodeSessionInventory, + decodeSessionPermissionMode, + decodeSessionTranscript, + encodeScheduleSpec, + encodeSessionPermissionMode, + parseMecatlEvent, + type ScheduleCarriedSpec, + type ScheduleFireRow, + type ScheduleRow, + type ScheduleSpecDraft, + type SessionInventoryPage, + type SessionPermissionMode, + type SessionSummary, + type SessionTranscript, + translateEvent, +} from "@/lib/protocol"; + +// The daemon API base: the proxy is transparent (no path rewriting), so the +// /v1 prefix belongs to the client's own URLs. Exported for the sibling +// watch module (watch.ts), which shares this transport's conventions. +export const HARNESS_API = "/api/mecatl/v1"; + +export interface HarnessStatus { + live: boolean; + detail: string; +} + +/** + * A daemon HTTP error, typed on the stable machine `code` from the RFC 9457 + * problem-details body (ADR 0248). The legacy top-level `error` key is still + * sent by every daemon, so `message` always carries the server's own words; + * `code` is "" against a pre-problem-details daemon. Flow control should + * branch on `code`, never on message prose. + */ +export class HarnessApiError extends Error { + readonly status: number; + readonly code: string; + constructor(status: number, code: string, message: string) { + super(message); + this.name = "HarnessApiError"; + this.status = status; + this.code = code; + } +} + +/** Codes whose raw detail deserves plainer user-facing framing. */ +const codeFraming: Record = { + draining: "The daemon is restarting — try again in a moment.", + session_leased_elsewhere: + "Another client is driving this chat right now — try again when its run finishes.", +}; + +/** Decodes a non-OK response into a HarnessApiError. Exported for watch.ts. */ +export async function apiError(response: Response): Promise { + const fallback = `${response.status} ${response.statusText}`; + try { + const body = (await response.json()) as { + error?: string; + code?: string; + detail?: string; + title?: string; + }; + const code = typeof body.code === "string" ? body.code : ""; + const message = + codeFraming[code] ?? body.error ?? body.detail ?? body.title ?? fallback; + return new HarnessApiError(response.status, code, message); + } catch { + return new HarnessApiError(response.status, "", fallback); + } +} + +async function readError(response: Response): Promise { + return (await apiError(response)).message; +} + +/** + * Cheap liveness probe. A deployed instance has no daemon on loopback, so this + * failing is the expected path there — callers fall back to mock behaviour. + */ +export async function probeHarness( + signal?: AbortSignal, +): Promise { + try { + const response = await fetch(`${HARNESS_API}/models`, { + signal, + cache: "no-store", + }); + if (!response.ok) return { live: false, detail: await readError(response) }; + return { live: true, detail: "connected" }; + } catch (error) { + return { + live: false, + detail: error instanceof Error ? error.message : String(error), + }; + } +} + +/** + * The daemon's compatibility document (GET /v1/compatibility, ADR 0248): + * the API major, the operator-enabled server capabilities (no probe session + * needed), the open feature registry, and the operator's deployment label. + * Returns null against an older daemon without the endpoint. + */ +export interface HarnessCompatibility { + apiMajor: number; + features: string[]; + capabilities: Record; + deployment: string; +} + +export async function fetchHarnessCompatibility( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/compatibility`, { + signal, + cache: "no-store", + }); + if (response.status === 404) return null; // pre-ADR-0248 daemon + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + api_major?: number; + features?: unknown; + capabilities?: Record; + deployment?: string; + }; + return { + apiMajor: typeof body.api_major === "number" ? body.api_major : 0, + features: Array.isArray(body.features) + ? body.features.filter((f): f is string => typeof f === "string") + : [], + capabilities: body.capabilities ?? {}, + deployment: body.deployment ?? "", + }; +} + +/** + * Creates a harness session. The workspace every file and shell tool is rooted + * at is injected by the proxy from MECATL_WORKSPACE, so it is deliberately not a + * parameter here — the browser never needs to know the server's paths. + * + * The create body's `mode` runs through the daemon's same modeFromString as + * POST /mode, so "accept_edits" is accepted at creation directly — no + * create-then-setMode dance is needed for an accept-edits draft. + */ +export async function createHarnessSession( + mode: "default" | "plan" | "accept_edits" = "default", + options?: { modelId?: string; providerId?: string; signal?: AbortSignal }, +): Promise { + const response = await fetch(`${HARNESS_API}/sessions`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + // model_id is protojson snake_case; omitted entirely on auto-routing so + // the daemon's own selection applies. The daemon requires provider_id + // whenever model_id is set (a bare model is ambiguous across providers). + body: JSON.stringify( + options?.modelId + ? { + mode, + model_id: options.modelId, + ...(options.providerId ? { provider_id: options.providerId } : {}), + } + : { mode }, + ), + signal: options?.signal, + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { session_id?: string }; + if (!body.session_id) throw new Error("harness returned no session id"); + return body.session_id; +} + +/** + * Continues an existing chat on a different model. The daemon fixes a + * session's provider/model at create, so a switch is a FORK: a new session + * seeded from the source's history (source_session_id carryover) on the + * picked model, renamed to the source's title. A running/awaiting source + * answers 412 (ThreadSourceBusyError). Model omitted = the daemon's own + * routing/default. + */ +export async function forkHarnessSessionToModel( + sourceSessionId: string, + model: { modelId: string; providerId: string } | null, + title: string, + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/sessions`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + mode: "default", + source_session_id: sourceSessionId, + ...(model + ? { model_id: model.modelId, provider_id: model.providerId } + : {}), + }), + signal, + }); + if (response.status === 412) { + throw new ThreadSourceBusyError((await apiError(response)).message); + } + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { session_id?: string }; + if (!body.session_id) throw new Error("harness returned no session id"); + if (title) { + try { + await renameHarnessSession(body.session_id, title); + } catch { + // Cosmetic; the forked session itself must not be lost to a rename. + } + } + return body.session_id; +} + +/** + * The parent session was mid-run: the daemon refuses to fork history from a + * running/awaiting source (HTTP 412). Thrown as its own type so the thread UI + * can say "wait for the current response" instead of a generic failure. + */ +export class ThreadSourceBusyError extends Error { + constructor(detail: string) { + super(detail || "The source session is still running."); + this.name = "ThreadSourceBusyError"; + } +} + +/** + * Creates the daemon session backing a message thread: a normal session whose + * conversation is seeded from the parent's history (source_session_id + * carryover), then renamed so the sidebar reads "Thread: …". The workspace is + * proxy-injected, exactly like createHarnessSession. A running/awaiting + * parent answers 412, surfaced as ThreadSourceBusyError. + */ +export async function createThreadHarnessSession( + parentSessionId: string, + title: string, + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/sessions`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + mode: "default", + source_session_id: parentSessionId, + }), + signal, + }); + if (response.status === 412) { + throw new ThreadSourceBusyError((await apiError(response)).message); + } + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { session_id?: string }; + if (!body.session_id) throw new Error("harness returned no session id"); + try { + await renameHarnessSession(body.session_id, title); + } catch { + // The rename is cosmetic (the sidebar label). The thread session itself + // is live and must not be lost to a failed rename. + } + return body.session_id; +} + +/** + * How long a live stream may go silent before Studio declares it dead. Two + * minutes comfortably exceeds a slow tool call's quiet stretch while still + * catching a daemon that went away without closing the socket. + */ +const STREAM_IDLE_TIMEOUT_MS = 120_000; + +async function readWithIdleTimeout(read: Promise): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((_, reject) => { + timer = setTimeout( + () => + reject(new Error("Mecatl stopped sending updates for two minutes.")), + STREAM_IDLE_TIMEOUT_MS, + ); + }); + try { + return await Promise.race([read, timeout]); + } finally { + clearTimeout(timer); + } +} + +/** + * Streams one prompt turn, invoking `onEvent` per translated event. + * + * The response is text/event-stream: frames are separated by a blank line and + * the payload rides a `data:` line, so partial frames must be buffered across + * reads rather than parsed per chunk. + * + * Two guards make a dead run fail loudly instead of hanging as "Done.": + * each read races the idle timeout, and a stream that closes without a + * terminal `result` frame throws — the daemon always ends a run with one. + */ +export interface PromptPart { + kind: "image" | "audio"; + mime_type: string; + /** Standard base64 (the daemon decodes JSON strings into bytes). */ + data: string; +} + +export async function streamHarnessPrompt( + sessionId: string, + text: string, + parts: PromptPart[], + onEvent: (event: StreamEvent) => void, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/prompt`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(parts.length > 0 ? { text, parts } : { text }), + signal, + }, + ); + await relayEventStream(response, sessionId, onEvent); +} + +/** + * Drives the failed-step retry endpoint (`POST /v1/sessions/{id}/retry`, + * ADR 0239): no request body — the daemon re-drives the recorded failed step + * itself, so nothing is re-sent — and the response is the run's SSE stream, + * relayed exactly like a prompt's. An ineligible session answers 409 with the + * stable code `failed_step_retry_ineligible` (a HarnessApiError here). + */ +export async function retryHarnessRun( + sessionId: string, + onEvent: (event: StreamEvent) => void, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/retry`, + { method: "POST", signal }, + ); + await relayEventStream(response, sessionId, onEvent); +} + +/** The shared prompt-shaped SSE relay: frame buffering, idle timeout, and the + * terminal-result guard — one discipline for /prompt and /retry. */ +async function relayEventStream( + response: Response, + sessionId: string, + onEvent: (event: StreamEvent) => void, +): Promise { + if (!response.ok) throw await apiError(response); + if (!response.body) throw new Error("harness returned no event stream"); + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let sawResult = false; + + while (true) { + const { value, done } = await readWithIdleTimeout(reader.read()); + if (done) break; + buffer += decoder.decode(value, { stream: true }); + const frames = buffer.split("\n\n"); + buffer = frames.pop() ?? ""; + for (const frame of frames) { + const line = frame + .split("\n") + .find((candidate) => candidate.startsWith("data:")); + if (!line) continue; + const payload = line.slice("data:".length).trim(); + if (!payload) continue; + let events: StreamEvent[]; + try { + events = translateEvent(parseMecatlEvent(payload), sessionId); + } catch { + // A frame Studio cannot decode is surfaced, never silently dropped. + onEvent({ + type: "notice", + text: "Mecatl sent a frame this Studio version could not decode.", + }); + continue; + } + for (const translated of events) { + if (translated.type === "run_result") sawResult = true; + onEvent(translated); + } + } + } + if (!sawResult) { + throw new Error( + "The connection closed before Mecatl returned a final result.", + ); + } +} + +export type HarnessApprovalVerdict = "allow_once" | "allow_always" | "deny"; + +/** + * Resolves a parked permission ask with the daemon's three-way verdict: + * allow_once, allow_always (persists a permission rule), or deny. + * + * `expectedRunId` (when known) scopes the verdict to ONE run (ADR 0249): the + * daemon refuses with 409 `stale_run_control` if that run already ended, so + * a stale approval dialog can never act on the session's NEXT run (e.g. a + * schedule fire into the same session). + */ +export async function respondToHarnessApproval( + sessionId: string, + askId: string, + verdict: HarnessApprovalVerdict, + expectedRunId?: string, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/approve`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + ask_id: askId, + verdict, + ...(expectedRunId ? { expected_run_id: expectedRunId } : {}), + }), + }, + ); + if (!response.ok) throw await apiError(response); +} + +// ── Session inventory / transcripts (daemon session store) ────────────────── + +/** One page of `GET /v1/sessions`. */ +async function fetchSessionInventoryPage( + cursor?: string, + pageSize = 100, + signal?: AbortSignal, +): Promise { + const query = new URLSearchParams({ page_size: String(pageSize) }); + if (cursor) query.set("cursor", cursor); + const response = await fetch(`${HARNESS_API}/sessions?${query}`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + return decodeSessionInventory(await response.json()); +} + +/** + * Walks the session inventory to completion, bounded so a pathological store + * cannot loop the UI forever. Only a COMPLETE walk may be used to conclude a + * session is gone — a partial page proves nothing about absent rows. + */ +export async function fetchAllSessions( + signal?: AbortSignal, + maxPages = 25, +): Promise<{ sessions: SessionSummary[]; complete: boolean }> { + const sessions: SessionSummary[] = []; + let cursor: string | undefined; + for (let page = 0; page < maxPages; page += 1) { + const result = await fetchSessionInventoryPage(cursor, 100, signal); + sessions.push(...result.sessions); + if (!result.nextCursor) return { sessions, complete: true }; + cursor = result.nextCursor; + } + return { sessions, complete: false }; +} + +/** + * Renames a session. Returns the daemon's clamped title echo, which the UI + * adopts rather than assuming its input survived unmodified. + */ +export async function renameHarnessSession( + sessionId: string, + title: string, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/rename`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ title }), + }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { title?: string }; + return body.title ?? title; +} + +/** + * Reads a session's current permission mode from its snapshot + * (`GET /v1/sessions/{id}` echoes the aggregate's mode string). + */ +export async function fetchHarnessSessionMode( + sessionId: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { mode?: string }; + return decodeSessionPermissionMode(body.mode); +} + +/** + * Changes a session's permission mode (POST /v1/sessions/{id}/mode, protojson + * snake_case body — modeled on renameHarnessSession). The response echoes the + * updated session; the echoed mode is returned so the UI adopts the daemon's + * word rather than assuming its input took. The daemon refuses a mid-turn + * change (the aggregate rejects it while running/awaiting), which surfaces + * here as a thrown error. + */ +export async function setHarnessSessionMode( + sessionId: string, + mode: SessionPermissionMode, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/mode`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ mode: encodeSessionPermissionMode(mode) }), + }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { mode?: string }; + return decodeSessionPermissionMode(body.mode); +} + +/** Physically deletes a session's snapshot and sidecars. */ +export async function deleteHarnessSession(sessionId: string): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/delete`, + { method: "POST" }, + ); + if (!response.ok) throw await apiError(response); +} + +/** + * Reads the authoritative message-level transcript + * (`GET /v1/sessions/{id}/transcript`). This is the store's snapshot, so it + * covers scheduler-tick fires whose conversation never reached the durable + * event log, and it works identically in external mode. + */ +export async function fetchSessionTranscriptMessages( + sessionId: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/transcript`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + return decodeSessionTranscript(await response.json()); +} + +// ── Memory (user model) ───────────────────────────────────────────────────── + +/** + * Reads the harness user model: durable facts the agent has stored about the + * operator, cross-project. The API returns the INDEX only — key plus one-line + * description — and never entry values, which the agent loads with RecallUser. + * + * There is no write endpoint: the agent curates memory through injection-scanned + * tool calls, so this is read-only by construction, not by choice. + */ +export interface HarnessUserModel { + entries: { key: string; description: string }[]; + /** Aggregate byte length of the rendered entries. */ + sizeBytes: number; + /** Lowercase-hex SHA-256 over the rendered entries — change detection. */ + sha256: string; +} + +export async function fetchHarnessUserModel( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/usermodel`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + entries?: { key?: string; description?: string }[]; + size_bytes?: number | string; + sha256?: string; + }; + return { + entries: (body.entries ?? []).map((entry) => ({ + key: entry.key ?? "", + description: entry.description ?? "", + })), + sizeBytes: Number(body.size_bytes ?? 0) || 0, + sha256: body.sha256 ?? "", + }; +} + +// ── Schedules ─────────────────────────────────────────────────────────────── + +/** + * Runs one schedule action. NOTE: "fire" is synchronous on the harness — the + * request stays open for the whole agent run, which can be minutes. + */ +export async function harnessScheduleAction( + name: string, + action: "pause" | "resume" | "fire" | "delete", +): Promise { + const encoded = encodeURIComponent(name); + const path = + action === "delete" + ? `${HARNESS_API}/schedules/${encoded}` + : `${HARNESS_API}/schedules/${encoded}/${action}`; + const response = await fetch(path, { + method: action === "delete" ? "DELETE" : "POST", + }); + if (!response.ok) throw await apiError(response); +} + +/** + * Reads the schedule registry through the full protocol decoder: spec fields + * (mode, mutating, workspace, timezone, limits), state, fire stage, and the + * carried spec an edit must round-trip. Distinguishes "scheduler not wired" + * (non-OK list — the daemon answers 501 without a schedule store) from a + * genuinely empty registry. + */ +export async function listScheduleRows( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/schedules`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + return decodeScheduleRows(await response.json()); +} + +/** + * Creates (POST) or replaces (PUT) a schedule spec. Requests are protojson — + * built by encodeScheduleSpec, never by echoing a decoded response — and an + * update must pass the row's `carried` spec or the fields this UI cannot edit + * would be silently deleted (PUT replaces the whole spec). + */ +export async function saveHarnessSchedule( + draft: ScheduleSpecDraft, + options: { update: boolean; carried?: ScheduleCarriedSpec }, +): Promise { + const body = JSON.stringify(encodeScheduleSpec(draft, options.carried)); + const response = await fetch( + options.update + ? `${HARNESS_API}/schedules/${encodeURIComponent(draft.name)}` + : `${HARNESS_API}/schedules`, + { + method: options.update ? "PUT" : "POST", + headers: { "Content-Type": "application/json" }, + body, + }, + ); + if (!response.ok) throw await apiError(response); +} + +/** Reads a schedule's fire history, newest first. */ +export async function listScheduleFires( + name: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/schedules/${encodeURIComponent(name)}/fires`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + return decodeScheduleFires(await response.json()); +} + +// ── Slash commands ─────────────────────────────────────────────────────────── + +/** + * Reads the workspace's discovered slash commands. The workspace query is + * injected by the proxy — the browser deliberately never knows the path. + */ +export async function listHarnessCommands( + signal?: AbortSignal, +): Promise<{ name: string; description: string }[]> { + const response = await fetch(`${HARNESS_API}/commands`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + commands?: { name?: string; description?: string }[]; + }; + return (body.commands ?? []).map((command) => ({ + name: command.name ?? "", + description: command.description ?? "", + })); +} + +// ── Skills ────────────────────────────────────────────────────────────────── + +/** + * Reads the skill inventory the daemon resolved at startup from its --skills-dir. + * The model sees only each skill's name and one-line summary until it chooses to + * load one, which is exactly what this returns. + */ +export interface HarnessSkillInfo { + name: string; + description: string; + /** Learned-lifecycle provenance; absent for immutable external skills. */ + agentOwned: boolean; + ownerAgent: string; + activeVersion: string; +} + +export async function listHarnessSkills( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/skills`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + skills?: { + name?: string; + description?: string; + agent_owned?: boolean; + owner_agent?: string; + active_version?: string; + }[]; + }; + return (body.skills ?? []).map((skill) => ({ + name: skill.name ?? "", + description: skill.description ?? "", + agentOwned: skill.agent_owned === true, + ownerAgent: skill.owner_agent ?? "", + activeVersion: skill.active_version ?? "", + })); +} + +/** Reads the selectable provider/model inventory. Carries no secret material. */ +export async function listHarnessModels(signal?: AbortSignal): Promise< + { + id: string; + providerId: string; + displayName: string; + contextLimit: number; + image: boolean; + reasoning: boolean; + }[] +> { + const response = await fetch(`${HARNESS_API}/models`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + models?: { + id?: string; + provider_id?: string; + display_name?: string; + context_limit?: number; + image?: boolean; + reasoning?: boolean; + }[]; + }; + return (body.models ?? []).map((model) => ({ + id: model.id ?? "", + providerId: model.provider_id ?? "", + displayName: model.display_name ?? model.id ?? "", + contextLimit: Number(model.context_limit ?? 0), + image: model.image === true, + reasoning: model.reasoning === true, + })); +} + +// ── Controller: provider, model router, MCP gateway ───────────────────────── + +const CONTROL_API = "/api/mecatl-control"; + +export interface HarnessControlStatus { + /** "external" when Studio proxies to MECATL_BASE_URL; "managed" otherwise. */ + mode: "managed" | "external"; + provider: string; + /** True when `provider` is the offline mock — the daemon serves canned + * turns rather than calling a real model. */ + isMock: boolean; + running: boolean; + gateway: { name: string; url: string } | null; + toolhiveGateway: { available: boolean; active: boolean } | null; + modelRouter: { enabled: boolean; categories: number } | null; + operatorSettings: boolean; + skillsDir: string; + memoryDir: string; + /** + * Provider NAMES found in the operator's auth.yaml — never credentials. + * Managed mode can now add/remove blocks THROUGH the controller (which + * owns the file server-side); no key value ever crosses this boundary. + */ + configuredProviders: string[]; + /** Which provider is active right now — "mock", "toolhive", or one of + * `configuredProviders` — seeded from MECATL_STUDIO_PROVIDER at startup + * but changeable at runtime via setActiveHarnessProvider. */ + selectedProvider: string | null; + /** The auth.yaml path on the controller's machine (guided-add copy). */ + authFile: string; +} + +export async function fetchHarnessControlStatus( + signal?: AbortSignal, +): Promise { + try { + const response = await fetch(`${CONTROL_API}/status`, { + signal, + cache: "no-store", + }); + if (!response.ok) return null; + const body = (await response.json()) as { + mode?: string; + provider?: string; + isMock?: boolean; + running?: boolean; + gateway?: { name?: string; url?: string } | null; + toolhiveGateway?: { available?: boolean; active?: boolean } | null; + modelRouter?: { enabled?: boolean; categories?: number } | null; + operatorSettings?: boolean; + skills?: { dir?: string }; + memory?: { dir?: string }; + configuredProviders?: unknown; + selectedProvider?: string | null; + authFile?: string; + }; + return { + mode: body.mode === "external" ? "external" : "managed", + provider: body.provider ?? "unknown", + isMock: Boolean(body.isMock), + running: Boolean(body.running), + gateway: body.gateway?.url + ? { name: body.gateway.name ?? "gateway", url: body.gateway.url } + : null, + toolhiveGateway: body.toolhiveGateway + ? { + available: Boolean(body.toolhiveGateway.available), + active: Boolean(body.toolhiveGateway.active), + } + : null, + modelRouter: body.modelRouter + ? { + enabled: Boolean(body.modelRouter.enabled), + categories: Number(body.modelRouter.categories ?? 0), + } + : null, + operatorSettings: Boolean(body.operatorSettings), + configuredProviders: Array.isArray(body.configuredProviders) + ? body.configuredProviders.filter( + (name): name is string => typeof name === "string", + ) + : [], + selectedProvider: body.selectedProvider ?? null, + authFile: body.authFile ?? "", + skillsDir: body.skills?.dir ?? "", + memoryDir: body.memory?.dir ?? "", + }; + } catch { + return null; + } +} + +export interface HarnessRouterCategory { + name: string; + description: string; + model: string; +} + +export interface HarnessRouterConfig { + enabled: boolean; + classifierModel: string; + defaultCategory: string; + categories: HarnessRouterCategory[]; + /** True when routing comes from an imported operator settings file, which this UI must not overwrite. */ + managedByOperator: boolean; +} + +export async function fetchHarnessRouter( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${CONTROL_API}/model-router`, { + signal, + cache: "no-store", + }); + if (!response.ok) return null; + const body = (await response.json()) as { + config?: { + enabled?: boolean; + classifierModel?: string; + defaultCategory?: string; + categories?: { name?: string; description?: string; model?: string }[]; + } | null; + managedBy?: string; + }; + return { + enabled: Boolean(body.config?.enabled), + classifierModel: body.config?.classifierModel ?? "", + defaultCategory: body.config?.defaultCategory ?? "", + categories: (body.config?.categories ?? []).map((category) => ({ + name: category.name ?? "", + description: category.description ?? "", + model: category.model ?? "", + })), + managedByOperator: body.managedBy === "operator-settings", + }; +} + +/** Saves routing config. RESTARTS the daemon. */ +export async function saveHarnessRouter( + config: Omit, +): Promise { + const response = await fetch(`${CONTROL_API}/model-router`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(config), + }); + if (!response.ok) throw await apiError(response); +} + +/** + * Connects an MCP gateway. RESTARTS the daemon, and a failed handshake rolls the + * previous gateway back on the controller side. + * + * The token is passed straight through to the loopback controller and is never + * stored, logged, or echoed by this UI. + */ +export async function connectHarnessGateway( + name: string, + url: string, + token?: string, +): Promise { + const response = await fetch(`${CONTROL_API}/mcp`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + name, + url, + token: token?.trim() || undefined, + }), + }); + if (!response.ok) throw await apiError(response); +} + +// ── Controller: provider management ───────────────────────────────────────── +// auth.yaml stays server-side property of the controller: these calls move +// NAMES and booleans, never key material. There is deliberately no +// "add provider with key" call — adding one is a guided copy-into-auth.yaml +// (see the Add-provider dialog), so a credential never transits the browser. + +/** One provider block found in the controller's auth.yaml — names and + * booleans only, never values (Studio rule 3). */ +export interface HarnessProviderInfo { + name: string; + configured: boolean; + /** A non-empty api_key / oauth access_token exists in the block. */ + keyPresent: boolean; + source: string; + /** The controller can key-test this kind with one cheap keyed call. */ + testable: boolean; +} + +export async function listHarnessProviders( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${CONTROL_API}/providers`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + providers?: { + name?: string; + configured?: boolean; + keyPresent?: boolean; + source?: string; + testable?: boolean; + }[]; + }; + return (body.providers ?? []) + .filter((row) => typeof row.name === "string" && row.name !== "") + .map((row) => ({ + name: row.name ?? "", + configured: row.configured !== false, + keyPresent: row.keyPresent === true, + source: row.source ?? "auth.yaml", + testable: row.testable === true, + })); +} + +/** One provider kind the daemon understands, with the guided-add snippet + * (a `` placeholder — never a real value). */ +export interface KnownHarnessProvider { + name: string; + label: string; + testable: boolean; + snippet: string; + note: string; +} + +export async function listKnownHarnessProviders( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${CONTROL_API}/providers/known`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + known?: { + name?: string; + label?: string; + testable?: boolean; + snippet?: string; + note?: string; + }[]; + }; + return (body.known ?? []) + .filter((row) => typeof row.name === "string" && row.name !== "") + .map((row) => ({ + name: row.name ?? "", + label: row.label ?? row.name ?? "", + testable: row.testable === true, + snippet: row.snippet ?? "", + note: row.note ?? "", + })); +} + +/** The controller's verdict on a stored key after ONE bounded probe. */ +export interface HarnessProviderKeyTest { + ok: boolean; + /** HTTP status from the provider (0 = unreachable/timeout). */ + status: number; + /** True when the provider answered 401/403 — the KEY is bad, not the wire. */ + rejected: boolean; + error: string; +} + +/** + * Asks the controller to test a provider's STORED key with one cheap + * authenticated call. The key itself never reaches the browser — only the + * verdict does. Throws when the test could not run at all (unknown provider, + * no key in auth.yaml, external mode's 409). + */ +export async function testHarnessProviderKey( + name: string, +): Promise { + const response = await fetch( + `${CONTROL_API}/providers/${encodeURIComponent(name)}/test`, + { method: "POST" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + ok?: boolean; + status?: number; + rejected?: boolean; + error?: string; + }; + return { + ok: body.ok === true, + status: Number(body.status ?? 0), + rejected: body.rejected === true, + error: body.error ?? "", + }; +} + +/** Removes a provider's block from auth.yaml. RESTARTS the daemon. */ +export async function removeHarnessProvider(name: string): Promise { + const response = await fetch( + `${CONTROL_API}/providers/${encodeURIComponent(name)}`, + { method: "DELETE" }, + ); + if (!response.ok) throw await apiError(response); +} + +/** Restarts the daemon with its current config — how a provider block just + * added to auth.yaml (guided add) becomes visible to mecated. */ +export async function restartHarnessDaemon(): Promise { + const response = await fetch(`${CONTROL_API}/restart`, { method: "POST" }); + if (!response.ok) throw await apiError(response); +} + +/** + * Switches the daemon's active provider — "mock", "toolhive", or any name + * already configured in auth.yaml — and restarts it on the spot. This is the + * live equivalent of setting MECATL_STUDIO_PROVIDER and restarting `npm run + * dev`: no credential travels with the request, only the chosen name. + */ +export async function setActiveHarnessProvider( + kind: string, +): Promise<{ provider: string; selectedProvider: string | null }> { + const response = await fetch(`${CONTROL_API}/providers/active`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ kind }), + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + provider?: string; + selectedProvider?: string | null; + }; + return { + provider: body.provider ?? "", + selectedProvider: body.selectedProvider ?? null, + }; +} + +// ── Controller: workspace skills management ───────────────────────────────── +// The daemon has no skill write API (its skills snapshot is resolved once at +// startup), so skill CRUD goes to the managed-mode controller, which owns the +// pinned --skills-dir and restarts mecated when a change touches what the +// daemon can see. External mode has no controller: every one of these answers +// 409 there, and the UI renders the controls disabled instead of calling them. + +/** A skill parked in the controller's `.disabled/` holding area. */ +export interface DisabledSkillInfo { + name: string; + description: string; +} + +/** Client-side mirror of the controller's name gate, so a bad name fails with + * this message rather than as a mystery 400. */ +function requireSkillName(name: string): string { + if (!validSkillName(name)) { + throw new Error( + "Skill names use lowercase letters, digits, hyphens, and underscores (max 64 characters)", + ); + } + return encodeURIComponent(name); +} + +/** Skills the controller has disabled (moved out of the daemon's sight). */ +export async function listDisabledHarnessSkills( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${CONTROL_API}/skills/disabled`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + disabled?: { name?: string; description?: string }[]; + }; + return (body.disabled ?? []) + .filter((skill) => typeof skill.name === "string" && skill.name !== "") + .map((skill) => ({ + name: skill.name ?? "", + description: skill.description ?? "", + })); +} + +/** Creates a new skill folder with its SKILL.md, enabled. RESTARTS the daemon + * — a new skill is invisible until the startup snapshot is rebuilt. */ +export async function createHarnessSkill( + name: string, + body: string, +): Promise { + requireSkillName(name); + const response = await fetch(`${CONTROL_API}/skills`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ name, body }), + }); + if (!response.ok) throw await apiError(response); +} + +/** Reads a skill's SKILL.md from the controller (either side of disabled). */ +export async function fetchHarnessSkillBody( + name: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${CONTROL_API}/skills/${requireSkillName(name)}/body`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { body?: string }; + return body.body ?? ""; +} + +/** One file of a multi-file skill create (a zip/folder upload). */ +export interface HarnessSkillUploadFile { + /** Relative POSIX path inside the skill folder, e.g. "scripts/run.sh". */ + path: string; + contentBase64: string; +} + +/** Creates a whole folder skill from an upload's files (must include a + * root SKILL.md). RESTARTS the daemon, like every skill mutation. */ +export async function createHarnessSkillFiles( + name: string, + files: HarnessSkillUploadFile[], +): Promise { + const response = await fetch(`${CONTROL_API}/skills`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ name: requireSkillName(name), files }), + }); + if (!response.ok) throw await apiError(response); +} + +/** One bundled file in a skill's folder. */ +export interface HarnessSkillFile { + /** Relative POSIX path inside the skill folder, e.g. "scripts/run.sh". */ + path: string; + size: number; +} + +/** Lists the files bundled in a skill's folder (works for disabled skills). */ +export async function listHarnessSkillFiles( + name: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${CONTROL_API}/skills/${requireSkillName(name)}/files`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { files?: HarnessSkillFile[] }; + return (body.files ?? []).filter( + (file) => typeof file?.path === "string" && typeof file?.size === "number", + ); +} + +/** Reads one bundled text file from a skill's folder (bounded server-side; + * binary or oversized files answer with a refusal that renders verbatim). */ +export async function fetchHarnessSkillFile( + name: string, + path: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${CONTROL_API}/skills/${requireSkillName(name)}/file?path=${encodeURIComponent(path)}`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { content?: string }; + return body.content ?? ""; +} + +/** Writes a skill's SKILL.md. RESTARTS the daemon when the skill is enabled. */ +export async function saveHarnessSkillBody( + name: string, + body: string, +): Promise { + const response = await fetch( + `${CONTROL_API}/skills/${requireSkillName(name)}/body`, + { + method: "PUT", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ body }), + }, + ); + if (!response.ok) throw await apiError(response); +} + +/** Moves a skill in or out of the disabled holding area. RESTARTS the daemon. */ +export async function setHarnessSkillEnabled( + name: string, + enabled: boolean, +): Promise { + const response = await fetch( + `${CONTROL_API}/skills/${requireSkillName(name)}/${enabled ? "enable" : "disable"}`, + { method: "POST" }, + ); + if (!response.ok) throw await apiError(response); +} + +/** Deletes a skill's directory from the workspace. RESTARTS the daemon. */ +export async function deleteHarnessSkill(name: string): Promise { + const response = await fetch( + `${CONTROL_API}/skills/${requireSkillName(name)}`, + { method: "DELETE" }, + ); + if (!response.ok) throw await apiError(response); +} + +/** One row of the daemon's resolved agent inventory (proto AgentInfo). */ +export interface HarnessAgentInfo { + name: string; + description: string; + /** Resolved pinned model id; empty = inherit ("auto": session model or routed). */ + model: string; + /** The def's effective read-only tool scope at the delegation call site. */ + tools: string[]; + /** Raw frontmatter permission mode; empty means "default". */ + permissionMode: string; + /** Optional UX color hint from the def; never affects execution. */ + color: string; +} + +/** Reads the daemon's RESOLVED agent inventory (what it can delegate to now). */ +export async function listHarnessAgents( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/agents`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + agents?: { + name?: string; + description?: string; + model?: string; + tools?: unknown; + permission_mode?: string; + color?: string; + }[]; + }; + return (body.agents ?? []).map((agent) => ({ + name: agent.name ?? "", + description: agent.description ?? "", + model: agent.model ?? "", + tools: Array.isArray(agent.tools) + ? agent.tools.filter((tool): tool is string => typeof tool === "string") + : [], + permissionMode: agent.permission_mode ?? "", + color: agent.color ?? "", + })); +} + +/** + * Begins the gateway's OAuth flow and returns the URL to send the user to. + * + * The controller performs discovery and dynamic client registration, then waits + * for the provider to redirect back to its own loopback callback, where it + * exchanges the code, stores the token and reconnects the daemon. + */ +export async function startHarnessGatewayOAuth( + name: string, + url: string, +): Promise { + const query = new URLSearchParams({ name, url }); + const response = await fetch(`${CONTROL_API}/mcp/oauth/start?${query}`); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { authorizationUrl?: string }; + if (!body.authorizationUrl) { + throw new Error("gateway returned no authorization URL"); + } + return body.authorizationUrl; +} + +/** + * Waits for the controller to report a connected gateway. + * + * Polling rather than listening for the callback page's postMessage: that + * message is addressed to a hardcoded origin (Studio's own port), so it never + * arrives here. The controller's status is the shared source of truth either way. + */ +export async function waitForHarnessGateway( + isCancelled: () => boolean, + timeoutMs = 180_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (isCancelled()) return false; + const status = await fetchHarnessControlStatus(); + if (status?.gateway) return true; + await new Promise((resolve) => setTimeout(resolve, 1_500)); + } + return false; +} + +/** + * Cancels the in-flight run for a session. `expectedRunId` (when known) + * scopes the cancel to ONE run (ADR 0249): a 409 `stale_run_control` means + * that run already ended — the desired outcome — and the session's NEXT run + * is left untouched. Best-effort either way: cancel is fire-and-forget. + */ +export async function cancelHarnessRun( + sessionId: string, + expectedRunId?: string, +): Promise { + await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/cancel`, + expectedRunId + ? { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ expected_run_id: expectedRunId }), + } + : { method: "POST" }, + ).catch(() => undefined); +} + +/** The daemon's verdict on a steer request. `accepted`/`appended` mean the + * text will be injected into the in-flight run at the next turn boundary; + * `too_late` means the run is past injection and the caller keeps the text + * (send it as a normal prompt instead). */ +export type HarnessSteerOutcome = "accepted" | "appended" | "too_late"; + +/** + * Injects a message into a session's in-flight run (ADR 0252). `messageId` is + * the client-minted correlation id: the stream's later `steer` echo names the + * id of the LAST message merged into the drained bundle, and the client + * splits its ordered pending list on that watermark. + * + * `expectedRunId` makes the steer STRICT (ADR 0249): the daemon refuses with + * 409 `stale_run_control` (a HarnessApiError here) when that run already + * ended, instead of PROMOTING the text into a fresh follow-up run behind + * Studio's back — the caller keeps the text and requeues it, exactly the + * too_late outcome. Studio always sends it; an unqualified steer's promotion + * relays a whole run as SSE on this response, which this JSON decode cannot + * carry. + * + * `parts` carries staged image attachments in the same validated wire shape + * as the prompt body (ADR 0251) — a media-only steer is legal. + */ +export async function steerHarnessRun( + sessionId: string, + text: string, + messageId: string, + options?: { expectedRunId?: string; parts?: PromptPart[] }, +): Promise<{ outcome: HarnessSteerOutcome; messageId: string }> { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/steer`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + text, + message_id: messageId, + ...(options?.parts?.length ? { parts: options.parts } : {}), + ...(options?.expectedRunId + ? { expected_run_id: options.expectedRunId } + : {}), + }), + }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + outcome?: string; + message_id?: string; + }; + // Anything other than an explicit accept resolves as too_late: the caller + // keeps the text and sends it as a normal prompt — never loses it. + const outcome: HarnessSteerOutcome = + body.outcome === "accepted" || body.outcome === "appended" + ? body.outcome + : "too_late"; + return { outcome, messageId: body.message_id ?? messageId }; +} + +/** + * Retracts the whole pending steer bundle (the daemon models one bundle per + * run, not per-message retraction). `none_pending` means nothing was waiting + * — the bundle already drained into the run or none was ever sent. + * + * ADR 0252 names the route `cancel-steer` (mirroring cancel-child); the + * pre-ADR `steer-cancel` spelling is a deprecated alias server-side. + */ +export async function cancelHarnessSteer( + sessionId: string, +): Promise<"retracted" | "none_pending"> { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/cancel-steer`, + { method: "POST" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { outcome?: string }; + return body.outcome === "retracted" ? "retracted" : "none_pending"; +} + +/** + * Manually compacts a session's conversation (`POST .../compact`, ADR 0244): + * bodyless request, `{"compacted": bool}` answer — true means the model + * history was rewritten (refetch the transcript), false means there was + * nothing to compact. 412 while the session is active/awaiting; gate the + * affordance on the `manual_compaction` capability. + */ +export async function compactHarnessSession( + sessionId: string, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/compact`, + { method: "POST" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { compacted?: boolean }; + return body.compacted === true; +} + +/** + * The provider+model a session actually resolved to (`GET /v1/sessions/{id}` + * echoes `resolved_model`, ADR 0244): the effective model label and the + * context window the context meter is measured against. Null when the daemon + * reports none (older daemon / unresolved model). + */ +export interface HarnessResolvedModel { + providerId: string; + modelId: string; + contextWindow: number; +} + +/** GET-session detail Studio consumes beyond the mode string. */ +export interface HarnessSessionDetail { + resolvedModel: HarnessResolvedModel | null; + /** The server capabilities echo when the daemon stamps one on the session + * response (B1.4); older daemons omit it — fall back to /compatibility. */ + capabilities: Record; +} + +export async function fetchHarnessSessionDetail( + sessionId: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { + resolved_model?: { + provider_id?: string; + model_id?: string; + context_window?: number | string; + } | null; + capabilities?: Record | null; + }; + const resolved = body.resolved_model; + return { + resolvedModel: resolved + ? { + providerId: resolved.provider_id ?? "", + modelId: resolved.model_id ?? "", + contextWindow: Number(resolved.context_window ?? 0) || 0, + } + : null, + capabilities: body.capabilities ?? {}, + }; +} diff --git a/studio/src/lib/harness/create-body.test.ts b/studio/src/lib/harness/create-body.test.ts new file mode 100644 index 0000000000..66ffab7c88 --- /dev/null +++ b/studio/src/lib/harness/create-body.test.ts @@ -0,0 +1,129 @@ +import { describe, expect, it, vi } from "vitest"; +import { + createHarnessSession, + createThreadHarnessSession, + forkHarnessSessionToModel, +} from "./client"; +import { createHarnessDebugSession } from "./debug"; + +/** + * Compatibility pin (requirement H4): the daemon strictly decodes the + * POST /v1/sessions create body — an unknown or mis-cased key (protojson + * camelCase included) is a 400 naming the field. These tests freeze the exact + * field sets Studio's create paths may emit so a stray key fails HERE, not as + * a baffling runtime 400. Extending the body is fine — extend the allowlist + * in the same change, knowing the daemon accepts the field. + */ + +const ALLOWED = new Set([ + "mode", + "model_id", + "provider_id", + "source_session_id", +]); + +/** + * The debug create (ADR 0254) is pinned as its OWN exact set rather than by + * widening ALLOWED: it is the one create that may carry `workspace` (and only + * as ""), `profile`, and the debug fields — the daemon accepts all of them — + * while the chat/thread/fork bodies must stay workspace-free (rule 2: the + * workspace is proxy-injected, never browser-supplied). Folding these keys + * into the shared allowlist would let a stray workspace on a PLAIN create + * pass this suite silently. + */ +const DEBUG_ALLOWED = [ + "debug_mcp_servers", + "debug_target_session_id", + "mode", + "profile", + "workspace", +] as const; + +function captureBody(): { body: () => Record } { + const captured: { value?: Record } = {}; + vi.stubGlobal("fetch", async (url: unknown, init?: RequestInit) => { + // Only the CREATE call is pinned — the thread/fork paths follow up with a + // cosmetic rename request whose body is a different contract. + if (String(url).endsWith("/v1/sessions")) { + captured.value = JSON.parse(String(init?.body ?? "{}")); + } + return new Response(JSON.stringify({ session_id: "s-1" }), { + status: 200, + }); + }); + return { + body: () => { + if (!captured.value) throw new Error("no create request captured"); + return captured.value; + }, + }; +} + +describe("session create bodies stay inside the daemon's strict field set", () => { + it("plain create with a model pick", async () => { + const captured = captureBody(); + try { + await createHarnessSession("plan", { + modelId: "m", + providerId: "openrouter", + }); + const keys = Object.keys(captured.body()); + expect(keys.every((k) => ALLOWED.has(k))).toBe(true); + // snake_case, never protojson camelCase. + expect(keys.some((k) => /[A-Z]/.test(k))).toBe(false); + } finally { + vi.unstubAllGlobals(); + } + }); + + it("auto-routed create omits the model fields entirely", async () => { + const captured = captureBody(); + try { + await createHarnessSession("default"); + expect(Object.keys(captured.body()).sort()).toEqual(["mode"]); + } finally { + vi.unstubAllGlobals(); + } + }); + + it("thread create", async () => { + const captured = captureBody(); + try { + await createThreadHarnessSession("parent-1", "Thread: x"); + const keys = Object.keys(captured.body()); + expect(keys.every((k) => ALLOWED.has(k))).toBe(true); + } finally { + vi.unstubAllGlobals(); + } + }); + + it("model-switch fork", async () => { + const captured = captureBody(); + try { + await forkHarnessSessionToModel( + "src-1", + { modelId: "m", providerId: "openrouter" }, + "Title", + ); + const keys = Object.keys(captured.body()); + expect(keys.every((k) => ALLOWED.has(k))).toBe(true); + } finally { + vi.unstubAllGlobals(); + } + }); + + it("debug create (ADR 0254) stays inside its own exact set", async () => { + const captured = captureBody(); + try { + await createHarnessDebugSession("target-1", { mcpServers: ["fetch"] }); + const body = captured.body(); + expect(Object.keys(body).sort()).toEqual([...DEBUG_ALLOWED]); + // The workspace this one create carries must be the EMPTY one the + // daemon requires — never a path. + expect(body.workspace).toBe(""); + expect(body.profile).toBe("no-fs"); + } finally { + vi.unstubAllGlobals(); + } + }); +}); diff --git a/studio/src/lib/harness/debug.test.ts b/studio/src/lib/harness/debug.test.ts new file mode 100644 index 0000000000..00b3858255 --- /dev/null +++ b/studio/src/lib/harness/debug.test.ts @@ -0,0 +1,73 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createHarnessDebugSession } from "./debug"; + +/** + * Pins the ADR-0254 debug-create wire contract: the daemon strictly decodes + * the body and REQUIRES profile "no-fs" with an EMPTY workspace alongside the + * target binding, so the exact field set (and the explicit `workspace: ""`) + * is load-bearing — a drift here is a runtime 400, not a cosmetic change. + */ + +type Captured = { url: string; init?: RequestInit }; + +function stubFetch(status: number, body: unknown): Captured { + const captured: Captured = { url: "" }; + vi.stubGlobal("fetch", async (url: RequestInfo | URL, init?: RequestInit) => { + captured.url = String(url); + captured.init = init; + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); + }); + return captured; +} + +afterEach(() => { + vi.unstubAllGlobals(); +}); + +describe("createHarnessDebugSession", () => { + it("sends exactly the ADR-0254 create body: no-fs profile, empty workspace, target", async () => { + const captured = stubFetch(201, { session_id: "dbg-1" }); + await expect(createHarnessDebugSession("target-1")).resolves.toBe("dbg-1"); + expect(captured.url).toBe("/api/mecatl/v1/sessions"); + expect(captured.init?.method).toBe("POST"); + const body = JSON.parse(String(captured.init?.body)); + expect(body).toEqual({ + mode: "default", + profile: "no-fs", + workspace: "", + debug_target_session_id: "target-1", + }); + // The empty workspace must be PRESENT, not merely falsy — it states the + // daemon's empty-workspace requirement on the wire. + expect(Object.hasOwn(body, "workspace")).toBe(true); + expect(body.workspace).toBe(""); + }); + + it("carries debug_mcp_servers only when servers are requested", async () => { + const captured = stubFetch(201, { session_id: "dbg-2" }); + await createHarnessDebugSession("target-1", { mcpServers: ["fetch"] }); + expect(JSON.parse(String(captured.init?.body))).toMatchObject({ + debug_target_session_id: "target-1", + debug_mcp_servers: ["fetch"], + }); + }); + + it("throws the typed error on a refusal (e.g. the 404 concealing ownership)", async () => { + stubFetch(404, { code: "not_found", error: "debug target is unavailable" }); + await expect(createHarnessDebugSession("target-x")).rejects.toMatchObject({ + name: "HarnessApiError", + status: 404, + code: "not_found", + }); + }); + + it("fails loudly when the daemon returns no session id", async () => { + stubFetch(201, {}); + await expect(createHarnessDebugSession("target-1")).rejects.toThrow( + "no session id", + ); + }); +}); diff --git a/studio/src/lib/harness/debug.ts b/studio/src/lib/harness/debug.ts new file mode 100644 index 0000000000..791d1b768c --- /dev/null +++ b/studio/src/lib/harness/debug.ts @@ -0,0 +1,56 @@ +/** + * AI session debugger — ADR 0254. + * + * Wire: `POST /v1/sessions` with `debug_target_session_id` creates a SEPARATE + * durable diagnostic session bound to one stored target. The daemon requires + * `profile: "no-fs"` and an EMPTY workspace (the debug engine has exactly one + * read-only tool, InspectSession, and never touches a filesystem), authorizes + * the target server-side before creating anything (an unowned or unknown + * target answers 404 — ownership failures are concealed as not-found), and + * never copies target conversation state. The new session then opens and runs + * like an ordinary chat; its inventory row carries + * `relationship.debug_target_session_id` plus capabilities that deny + * rename/delete. Gated by `capabilities.session_debug` (and the optional + * server list by `capabilities.debug_mcp`). + * + * CONSENT: invoking the debugger sends the target's stored transcript and + * event evidence — everything in it, secrets included — to the selected model. + * Callers must put the ADR-0254 disclosure in front of the user BEFORE calling + * this; this module only speaks the wire. + * + * The `workspace: ""` is explicit rather than omitted: it states the daemon's + * empty-workspace contract on the wire, and the server proxy's workspace + * injection (src/lib/server-proxy.ts) skips no-fs/debug creates so the empty + * value actually survives to the daemon. + */ + +import { apiError, HARNESS_API } from "./client"; + +/** Creates a debug session bound to `targetSessionId`; returns the new id. */ +export async function createHarnessDebugSession( + targetSessionId: string, + options?: { + /** Optional debug MCP server names (gated by `capabilities.debug_mcp`). */ + mcpServers?: string[]; + signal?: AbortSignal; + }, +): Promise { + const response = await fetch(`${HARNESS_API}/sessions`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + mode: "default", + profile: "no-fs", + workspace: "", + debug_target_session_id: targetSessionId, + ...(options?.mcpServers?.length + ? { debug_mcp_servers: options.mcpServers } + : {}), + }), + signal: options?.signal, + }); + if (!response.ok) throw await apiError(response); + const body = (await response.json()) as { session_id?: string }; + if (!body.session_id) throw new Error("harness returned no session id"); + return body.session_id; +} diff --git a/studio/src/lib/harness/dream.test.ts b/studio/src/lib/harness/dream.test.ts new file mode 100644 index 0000000000..0fc3c2901b --- /dev/null +++ b/studio/src/lib/harness/dream.test.ts @@ -0,0 +1,161 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { HarnessApiError } from "./client"; +import { + decideDreamPlan, + decodeDreamPlan, + dreamTargetCapability, + generateDreamPlan, + isStaleDreamPlan, +} from "./dream"; + +/** + * Pins the manual-dream wire contract (ADR 0227): the stdlib-JSON plan and + * receipt shapes, the process-local plan-id staleness classification + * (regenerate, never retry), and the capability-object reader. + */ + +const originalFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = originalFetch; +}); + +function respond( + status: number, + body: unknown, + capture?: { url?: string; init?: RequestInit }, +) { + globalThis.fetch = async (input, init) => { + if (capture) { + capture.url = String(input); + capture.init = init; + } + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); + }; +} + +describe("decodeDreamPlan", () => { + it("decodes operations with survivor/sources/replacement", () => { + const plan = decodeDreamPlan({ + id: "plan1", + target: "user_model", + expires_at: { seconds: 1788287318, nanos: 927881000 }, + planned_operation_count: 1, + planned_source_count: 2, + operations: [ + { + kind: "merge", + survivor: { key: "k1", value: "v1", description: "d1" }, + sources: [{ key: "k2", value: "v2" }], + replacement: { value: "merged", description: "both" }, + reason: "near-duplicates", + exact_duplicate_eligible: true, + }, + ], + }); + expect(plan.id).toBe("plan1"); + expect(plan.expiresAtUnix).toBe(1788287318); + expect(plan.plannedOperationCount).toBe(1); + expect(plan.operations).toHaveLength(1); + expect(plan.operations[0]).toMatchObject({ + kind: "merge", + survivor: { key: "k1", value: "v1", description: "d1" }, + replacement: { value: "merged", description: "both" }, + exactDuplicateEligible: true, + }); + expect(plan.operations[0].sources).toEqual([ + { key: "k2", value: "v2", description: "" }, + ]); + }); + + it("decodes the daemon's empty plan (nothing to consolidate)", () => { + const plan = decodeDreamPlan({ id: "p", target: "project_memory" }); + expect(plan.operations).toEqual([]); + expect(plan.plannedOperationCount).toBe(0); + }); +}); + +describe("generate / decide", () => { + it("posts the target and unwraps the plan", async () => { + const capture: { url?: string; init?: RequestInit } = {}; + respond(200, { plan: { id: "p1", target: "project_memory" } }, capture); + const plan = await generateDreamPlan("project_memory"); + expect(capture.url).toContain("/dream/plans"); + expect(JSON.parse(String(capture.init?.body))).toEqual({ + target: "project_memory", + }); + expect(plan.id).toBe("p1"); + }); + + it("posts the decision and decodes the receipt counts", async () => { + const capture: { url?: string; init?: RequestInit } = {}; + respond( + 200, + { + receipt: { + id: "p1", + target: "project_memory", + disposition: "apply", + planned_source_count: 3, + applied_source_count: 2, + conflicted_source_count: 1, + }, + }, + capture, + ); + const receipt = await decideDreamPlan("p1", "apply"); + expect(capture.url).toContain("/dream/plans/p1/decision"); + expect(JSON.parse(String(capture.init?.body))).toEqual({ + decision: "apply", + }); + expect(receipt).toMatchObject({ + disposition: "apply", + planned: 3, + applied: 2, + conflicted: 1, + skipped: 0, + failed: 0, + }); + }); + + it("classifies a dead plan id as stale (regenerate, never retry)", async () => { + respond(404, { + code: "dream_not_found", + error: "dream plan not found; generate a new plan", + }); + const error = await decideDreamPlan("stale", "apply").catch((e) => e); + expect(error).toBeInstanceOf(HarnessApiError); + expect(isStaleDreamPlan(error)).toBe(true); + expect( + isStaleDreamPlan(new HarnessApiError(410, "dream_terminal_conflict", "")), + ).toBe(true); + // In-progress/conflict are NOT stale: the plan still exists. + expect( + isStaleDreamPlan(new HarnessApiError(409, "dream_in_progress", "")), + ).toBe(false); + expect(isStaleDreamPlan(new Error("network"))).toBe(false); + }); +}); + +describe("dreamTargetCapability", () => { + it("reads the per-target object and defaults absent to all-false", () => { + const manualDream = { + project_memory: { generate: true, decide: true }, + user_model: { generate: false, decide: false, unavailable_reason: "off" }, + }; + expect(dreamTargetCapability(manualDream, "project_memory")).toEqual({ + generate: true, + decide: true, + unavailableReason: "", + }); + expect(dreamTargetCapability(manualDream, "user_model")).toEqual({ + generate: false, + decide: false, + unavailableReason: "off", + }); + expect(dreamTargetCapability(undefined, "user_model").generate).toBe(false); + expect(dreamTargetCapability(true, "project_memory").decide).toBe(false); + }); +}); diff --git a/studio/src/lib/harness/dream.ts b/studio/src/lib/harness/dream.ts new file mode 100644 index 0000000000..5c99dc70cb --- /dev/null +++ b/studio/src/lib/harness/dream.ts @@ -0,0 +1,178 @@ +/** + * Manual memory consolidation ("dream") review — ADR 0227. + * + * Wire: `POST /v1/dream/plans {target}` generates a bounded, daemon-curated + * consolidation plan; `POST /v1/dream/plans/{plan_id}/decision {decision}` + * applies or dismisses the WHOLE plan. Studio never composes memory content — + * the user only approves what the daemon curated (memory rule 8). + * + * Plan ids are process-local: a daemon restart (or the retention window + * expiring) answers `dream_not_found` (404) — regenerate, never retry. Gated + * by `capabilities.manual_dream.{project_memory,user_model}.{generate,decide}` + * on GET /v1/compatibility. + */ + +import { apiError, HARNESS_API, HarnessApiError } from "./client"; +import { + asArray, + asBool, + asNumber, + asRecord, + asString, + timestampUnix, +} from "./wire"; + +export type DreamTarget = "project_memory" | "user_model"; +export type DreamDecision = "apply" | "dismiss"; + +interface DreamParticipant { + key: string; + value: string; + description: string; +} + +interface DreamOperation { + kind: string; + survivor: DreamParticipant; + sources: DreamParticipant[]; + replacement: { value: string; description: string }; + reason: string; + exactDuplicateEligible: boolean; +} + +export interface DreamPlan { + id: string; + target: string; + expiresAtUnix: number; + plannedOperationCount: number; + plannedSourceCount: number; + operations: DreamOperation[]; +} + +export interface DreamReceipt { + id: string; + target: string; + disposition: string; + planned: number; + applied: number; + conflicted: number; + skipped: number; + failed: number; +} + +function decodeParticipant(raw: unknown): DreamParticipant { + const record = asRecord(raw); + return { + key: asString(record.key), + value: asString(record.value), + description: asString(record.description), + }; +} + +export function decodeDreamPlan(raw: unknown): DreamPlan { + const record = asRecord(raw); + return { + id: asString(record.id), + target: asString(record.target), + expiresAtUnix: timestampUnix(record.expires_at), + plannedOperationCount: asNumber(record.planned_operation_count), + plannedSourceCount: asNumber(record.planned_source_count), + operations: asArray(record.operations).map((operation) => { + const entry = asRecord(operation); + const replacement = asRecord(entry.replacement); + return { + kind: asString(entry.kind), + survivor: decodeParticipant(entry.survivor), + sources: asArray(entry.sources).map(decodeParticipant), + replacement: { + value: asString(replacement.value), + description: asString(replacement.description), + }, + reason: asString(entry.reason), + exactDuplicateEligible: asBool(entry.exact_duplicate_eligible), + }; + }), + }; +} + +function decodeDreamReceipt(raw: unknown): DreamReceipt { + const record = asRecord(raw); + return { + id: asString(record.id), + target: asString(record.target), + disposition: asString(record.disposition), + planned: asNumber(record.planned_source_count), + applied: asNumber(record.applied_source_count), + conflicted: asNumber(record.conflicted_source_count), + skipped: asNumber(record.skipped_source_count), + failed: asNumber(record.failed_source_count), + }; +} + +/** Generates a consolidation plan. Synchronous and potentially slow. */ +export async function generateDreamPlan( + target: DreamTarget, + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/dream/plans`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ target }), + signal, + }); + if (!response.ok) throw await apiError(response); + return decodeDreamPlan(asRecord(await response.json()).plan); +} + +/** Applies or dismisses the whole retained plan. */ +export async function decideDreamPlan( + planId: string, + decision: DreamDecision, +): Promise { + const response = await fetch( + `${HARNESS_API}/dream/plans/${encodeURIComponent(planId)}/decision`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ decision }), + }, + ); + if (!response.ok) throw await apiError(response); + return decodeDreamReceipt(asRecord(await response.json()).receipt); +} + +/** + * True when the plan id no longer resolves — unknown, expired, or minted by a + * daemon process that has since restarted. The fix is regenerate, not retry. + */ +export function isStaleDreamPlan(error: unknown): boolean { + return ( + error instanceof HarnessApiError && + (error.code === "dream_not_found" || + error.code === "dream_terminal_conflict" || + (error.code === "" && (error.status === 404 || error.status === 410))) + ); +} + +/** The per-target capability object off `capabilities.manual_dream`. */ +export interface DreamTargetCapability { + generate: boolean; + decide: boolean; + unavailableReason: string; +} + +/** + * Reads one target's capability out of the compatibility document's + * `manual_dream` object (absent daemon/target → all-false). + */ +export function dreamTargetCapability( + manualDream: unknown, + target: DreamTarget, +): DreamTargetCapability { + const entry = asRecord(asRecord(manualDream)[target]); + return { + generate: asBool(entry.generate), + decide: asBool(entry.decide), + unavailableReason: asString(entry.unavailable_reason), + }; +} diff --git a/studio/src/lib/harness/learned-skills.test.ts b/studio/src/lib/harness/learned-skills.test.ts new file mode 100644 index 0000000000..f15754fc2e --- /dev/null +++ b/studio/src/lib/harness/learned-skills.test.ts @@ -0,0 +1,185 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { + decodeLearnedSkillChange, + decodeLearnedSkillVersion, + diffLearnedSkillVersions, + listLearnedSkillChanges, + listLearnedSkills, + mutateLearnedSkill, + rollbackLearnedSkill, +} from "./learned-skills"; + +/** + * Pins the learned-skill wire contract (ADR 0110): the stdlib-JSON proto + * shapes, the owner_agent/version/expected_revision mutation body, and the + * rollback target_version body. + */ + +const originalFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = originalFetch; +}); + +function respond( + status: number, + body: unknown, + capture?: { url?: string; init?: RequestInit }, +) { + globalThis.fetch = async (input, init) => { + if (capture) { + capture.url = String(input); + capture.init = init; + } + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); + }; +} + +describe("decodeLearnedSkillVersion", () => { + it("decodes the snake_case projection", () => { + const skill = decodeLearnedSkillVersion({ + id: "sk1", + name: "triage-flakes", + version: "v2", + revision: "r7", + state: "staged", + owner_agent: "explorer", + description: "Triage flaky tests", + body: "# Steps", + supersedes: "v1", + evidence_count: 3, + updated_at: { seconds: 1788000001 }, + inspect_available: true, + }); + expect(skill).toMatchObject({ + id: "sk1", + name: "triage-flakes", + version: "v2", + revision: "r7", + state: "staged", + ownerAgent: "explorer", + supersedes: "v1", + evidenceCount: 3, + updatedAtUnix: 1788000001, + createdAtUnix: 0, + inspectAvailable: true, + undoAvailable: false, + }); + }); + + it("never throws on a foreign shape", () => { + expect(decodeLearnedSkillVersion(undefined).state).toBe(""); + expect(decodeLearnedSkillChange(42).operation).toBe(""); + }); +}); + +describe("list endpoints", () => { + it("treats the daemon's generation-only answer as empty inventories", async () => { + respond(200, { generation: 1 }); + expect(await listLearnedSkills()).toEqual({ skills: [], nextCursor: "" }); + respond(200, { generation: 1 }); + expect(await listLearnedSkillChanges()).toEqual({ + changes: [], + nextCursor: "", + }); + }); + + it("passes the state filter and cursor through", async () => { + const capture: { url?: string } = {}; + respond(200, { skills: [{ id: "sk1" }], next_cursor: "n" }, capture); + const page = await listLearnedSkills({ state: "staged", cursor: "c" }); + expect(capture.url).toContain("/skills/learned?"); + expect(capture.url).toContain("state=staged"); + expect(capture.url).toContain("cursor=c"); + expect(page.skills).toHaveLength(1); + expect(page.nextCursor).toBe("n"); + }); + + it("throws the typed error on a problem response", async () => { + respond(400, { code: "invalid_argument", error: "invalid limit" }); + await expect(listLearnedSkills({ limit: -1 })).rejects.toMatchObject({ + name: "HarnessApiError", + code: "invalid_argument", + }); + }); +}); + +describe("diffLearnedSkillVersions", () => { + it("queries owner_agent/from/to and returns the diff text", async () => { + const capture: { url?: string } = {}; + respond( + 200, + { diff: "-a\n+b", from_version: "v1", to_version: "v2" }, + capture, + ); + const diff = await diffLearnedSkillVersions("sk1", "explorer", "v1", "v2"); + expect(capture.url).toContain("/skills/learned/sk1/diff?"); + expect(capture.url).toContain("owner_agent=explorer"); + expect(capture.url).toContain("from=v1"); + expect(capture.url).toContain("to=v2"); + expect(diff).toBe("-a\n+b"); + }); +}); + +describe("mutations", () => { + it("posts owner_agent + version + expected_revision to the action route", async () => { + const capture: { url?: string; init?: RequestInit } = {}; + respond( + 200, + { + skill: { id: "sk1", state: "active" }, + publication_status: "published", + }, + capture, + ); + const result = await mutateLearnedSkill("activate", { + id: "sk1", + ownerAgent: "explorer", + version: "v2", + expectedRevision: "r7", + }); + expect(capture.url).toContain("/skills/learned/sk1/activate"); + expect(JSON.parse(String(capture.init?.body))).toEqual({ + owner_agent: "explorer", + version: "v2", + expected_revision: "r7", + }); + expect(result.skill.state).toBe("active"); + expect(result.publicationStatus).toBe("published"); + }); + + it("rollback posts target_version instead of version", async () => { + const capture: { url?: string; init?: RequestInit } = {}; + respond( + 200, + { skill: { id: "sk1", state: "active", version: "v1" } }, + capture, + ); + await rollbackLearnedSkill({ + id: "sk1", + ownerAgent: "explorer", + targetVersion: "v1", + expectedRevision: "r7", + }); + expect(capture.url).toContain("/skills/learned/sk1/rollback"); + expect(JSON.parse(String(capture.init?.body))).toEqual({ + owner_agent: "explorer", + target_version: "v1", + expected_revision: "r7", + }); + }); + + it("surfaces a not-found mutation as the typed error", async () => { + respond(404, { code: "session_not_found", error: "learned skill" }); + await expect( + mutateLearnedSkill("reject", { + id: "gone", + ownerAgent: "explorer", + version: "v1", + expectedRevision: "r1", + }), + ).rejects.toMatchObject({ name: "HarnessApiError", status: 404 }); + }); +}); diff --git a/studio/src/lib/harness/learned-skills.ts b/studio/src/lib/harness/learned-skills.ts new file mode 100644 index 0000000000..1a789a8960 --- /dev/null +++ b/studio/src/lib/harness/learned-skills.ts @@ -0,0 +1,244 @@ +/** + * Learned-skill lifecycle (ADR 0110) — the human half of the daemon's + * self-improvement loop. + * + * Wire: `GET /v1/skills/learned` (+ `/changes`, `/{id}`, `/{id}/diff`) and + * `POST /v1/skills/learned/{id}/{activate|reject|archive|rollback}`. Gated by + * `capabilities.learned_skills` on GET /v1/compatibility. + * + * Responses are stdlib JSON over the proto structs (snake_case keys, absent = + * zero value). Every mutation carries the version being acted on plus + * `expected_revision` — the optimistic-concurrency token off the listed row. + * The `project` query is deliberately never sent: Studio reads the + * operator-scope partition (the browser never knows the workspace path). + */ + +import { apiError, HARNESS_API } from "./client"; +import { + asArray, + asBool, + asNumber, + asRecord, + asString, + timestampUnix, +} from "./wire"; + +/** + * One immutable agent-owned skill version. `state` is the daemon's closed + * vocabulary: draft, evaluated, staged, active, archived, rejected. + */ +export interface LearnedSkillVersion { + id: string; + name: string; + version: string; + /** Optimistic-concurrency token — every mutation must carry it. */ + revision: string; + state: string; + ownerAgent: string; + description: string; + /** Bounded, server-repaired body preview. */ + body: string; + /** The prior version this one replaced ("" for a first version). */ + supersedes: string; + evidenceCount: number; + createdAtUnix: number; + updatedAtUnix: number; + inspectAvailable: boolean; + undoAvailable: boolean; +} + +export function decodeLearnedSkillVersion(raw: unknown): LearnedSkillVersion { + const record = asRecord(raw); + return { + id: asString(record.id), + name: asString(record.name), + version: asString(record.version), + revision: asString(record.revision), + state: asString(record.state), + ownerAgent: asString(record.owner_agent), + description: asString(record.description), + body: asString(record.body), + supersedes: asString(record.supersedes), + evidenceCount: asNumber(record.evidence_count), + createdAtUnix: timestampUnix(record.created_at), + updatedAtUnix: timestampUnix(record.updated_at), + inspectAvailable: asBool(record.inspect_available), + undoAvailable: asBool(record.undo_available), + }; +} + +/** One append-only lifecycle receipt from `GET /v1/skills/learned/changes`. */ +export interface LearnedSkillChange { + id: string; + skillId: string; + name: string; + version: string; + operation: string; + fromState: string; + toState: string; + evidenceCount: number; + verdict: string; + atUnix: number; +} + +export function decodeLearnedSkillChange(raw: unknown): LearnedSkillChange { + const record = asRecord(raw); + return { + id: asString(record.id), + skillId: asString(record.skill_id), + name: asString(record.name), + version: asString(record.version), + operation: asString(record.operation), + fromState: asString(record.from_state), + toState: asString(record.to_state), + evidenceCount: asNumber(record.evidence_count), + verdict: asString(record.verdict), + atUnix: timestampUnix(record.at), + }; +} + +export interface LearnedSkillPage { + skills: LearnedSkillVersion[]; + nextCursor: string; +} + +export async function listLearnedSkills( + options: { state?: string; cursor?: string; limit?: number } = {}, + signal?: AbortSignal, +): Promise { + const query = new URLSearchParams(); + if (options.state) query.set("state", options.state); + if (options.cursor) query.set("cursor", options.cursor); + if (options.limit) query.set("limit", String(options.limit)); + const suffix = query.size > 0 ? `?${query}` : ""; + const response = await fetch(`${HARNESS_API}/skills/learned${suffix}`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = asRecord(await response.json()); + return { + skills: asArray(body.skills).map(decodeLearnedSkillVersion), + nextCursor: asString(body.next_cursor), + }; +} + +export async function listLearnedSkillChanges( + options: { cursor?: string; limit?: number } = {}, + signal?: AbortSignal, +): Promise<{ changes: LearnedSkillChange[]; nextCursor: string }> { + const query = new URLSearchParams(); + if (options.cursor) query.set("cursor", options.cursor); + if (options.limit) query.set("limit", String(options.limit)); + const suffix = query.size > 0 ? `?${query}` : ""; + const response = await fetch( + `${HARNESS_API}/skills/learned/changes${suffix}`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + const body = asRecord(await response.json()); + return { + changes: asArray(body.changes).map(decodeLearnedSkillChange), + nextCursor: asString(body.next_cursor), + }; +} + +/** Reads one version in full (the list body is a bounded preview). */ +export async function fetchLearnedSkill( + id: string, + ownerAgent: string, + version?: string, + signal?: AbortSignal, +): Promise { + const query = new URLSearchParams({ owner_agent: ownerAgent }); + if (version) query.set("version", version); + const response = await fetch( + `${HARNESS_API}/skills/learned/${encodeURIComponent(id)}?${query}`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + return decodeLearnedSkillVersion(asRecord(await response.json()).skill); +} + +/** Unified diff between two versions of one learned skill. */ +export async function diffLearnedSkillVersions( + id: string, + ownerAgent: string, + fromVersion: string, + toVersion: string, + signal?: AbortSignal, +): Promise { + const query = new URLSearchParams({ + owner_agent: ownerAgent, + from: fromVersion, + to: toVersion, + }); + const response = await fetch( + `${HARNESS_API}/skills/learned/${encodeURIComponent(id)}/diff?${query}`, + { signal, cache: "no-store" }, + ); + if (!response.ok) throw await apiError(response); + return asString(asRecord(await response.json()).diff); +} + +export type LearnedSkillAction = "activate" | "reject" | "archive"; + +export interface LearnedSkillMutationResult { + skill: LearnedSkillVersion; + /** How republishing to the live skill snapshot went, when it applies. */ + publicationStatus: string; + publicationError: string; +} + +async function learnedSkillMutation( + id: string, + action: string, + body: Record, +): Promise { + const response = await fetch( + `${HARNESS_API}/skills/learned/${encodeURIComponent(id)}/${action}`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(body), + }, + ); + if (!response.ok) throw await apiError(response); + const decoded = asRecord(await response.json()); + return { + skill: decodeLearnedSkillVersion(decoded.skill), + publicationStatus: asString(decoded.publication_status), + publicationError: asString(decoded.publication_error), + }; +} + +/** Activate / reject / archive one version, guarded by its revision. */ +export async function mutateLearnedSkill( + action: LearnedSkillAction, + target: { + id: string; + ownerAgent: string; + version: string; + expectedRevision: string; + }, +): Promise { + return learnedSkillMutation(target.id, action, { + owner_agent: target.ownerAgent, + version: target.version, + expected_revision: target.expectedRevision, + }); +} + +/** Rolls the skill back to a prior version (usually `supersedes`). */ +export async function rollbackLearnedSkill(target: { + id: string; + ownerAgent: string; + targetVersion: string; + expectedRevision: string; +}): Promise { + return learnedSkillMutation(target.id, "rollback", { + owner_agent: target.ownerAgent, + target_version: target.targetVersion, + expected_revision: target.expectedRevision, + }); +} diff --git a/studio/src/lib/harness/learning.test.ts b/studio/src/lib/harness/learning.test.ts new file mode 100644 index 0000000000..5210c097e0 --- /dev/null +++ b/studio/src/lib/harness/learning.test.ts @@ -0,0 +1,190 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { HarnessApiError } from "./client"; +import { + decideLearningProposal, + decodeLearningProposal, + isProposalConflict, + listLearningProposals, + reflectHarnessSession, +} from "./learning"; + +/** + * Pins the learning-review wire contract (ADR 0109): stdlib-JSON proto shapes + * (snake_case keys, `{seconds}` timestamps, absent = zero value), the + * expected_version concurrency token on decisions, and the 409 + * proposal_conflict classification. + */ + +const originalFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = originalFetch; +}); + +function respond( + status: number, + body: unknown, + capture?: { url?: string; init?: RequestInit }, +) { + globalThis.fetch = async (input, init) => { + if (capture) { + capture.url = String(input); + capture.init = init; + } + return new Response(JSON.stringify(body), { + status, + headers: { "Content-Type": "application/json" }, + }); + }; +} + +describe("decodeLearningProposal", () => { + it("decodes the daemon's snake_case digest, tolerating absent fields", () => { + const proposal = decodeLearningProposal({ + id: "p1", + version: "v3", + status: "staged", + kind: "fact", + key: "deploy/steps", + title: "Deploy steps", + description: "How this repo deploys", + evidence: [{ session_id: "s1" }, { session_id: "s2" }], + triggers: ["deploy", 7], + decisions: [ + { kind: "approve", actor: "operator", at: { seconds: 1788000000 } }, + ], + created_at: { seconds: 1787000000, nanos: 5 }, + promotion_available: true, + learned_skill_id: "sk1", + }); + expect(proposal).toMatchObject({ + id: "p1", + version: "v3", + status: "staged", + key: "deploy/steps", + evidenceCount: 2, + triggers: ["deploy"], + createdAtUnix: 1787000000, + updatedAtUnix: 0, + promotionAvailable: true, + promotionUnavailableReason: "", + learnedSkillId: "sk1", + }); + expect(proposal.decisions).toEqual([ + { kind: "approve", actor: "operator", reason: "", atUnix: 1788000000 }, + ]); + }); + + it("degrades a completely foreign shape to zero values, never throws", () => { + expect(decodeLearningProposal(null).id).toBe(""); + expect(decodeLearningProposal("nope").triggers).toEqual([]); + expect(decodeLearningProposal({ decisions: "x" }).decisions).toEqual([]); + }); +}); + +describe("listLearningProposals", () => { + it("builds the status/cursor/limit query and decodes the page", async () => { + const capture: { url?: string } = {}; + respond(200, { proposals: [{ id: "p1" }], next_cursor: "c2" }, capture); + const page = await listLearningProposals({ + status: "staged", + cursor: "c1", + limit: 25, + }); + expect(capture.url).toContain("/learning/proposals?"); + expect(capture.url).toContain("status=staged"); + expect(capture.url).toContain("cursor=c1"); + expect(capture.url).toContain("limit=25"); + expect(page.proposals.map((p) => p.id)).toEqual(["p1"]); + expect(page.nextCursor).toBe("c2"); + }); + + it("treats the daemon's empty `{}` answer as an empty queue", async () => { + respond(200, {}); + const page = await listLearningProposals(); + expect(page).toEqual({ proposals: [], nextCursor: "" }); + }); + + it("throws the typed error on a problem response", async () => { + respond(501, { + code: "learning_unavailable", + error: "learning proposals are not configured", + }); + await expect(listLearningProposals()).rejects.toMatchObject({ + name: "HarnessApiError", + code: "learning_unavailable", + status: 501, + }); + }); +}); + +describe("decideLearningProposal", () => { + it("posts the decision with expected_version", async () => { + const capture: { url?: string; init?: RequestInit } = {}; + respond(200, { proposal: { id: "p1", status: "promoted" } }, capture); + const updated = await decideLearningProposal("p1", "approve", "v3"); + expect(capture.url).toContain("/learning/proposals/p1/decision"); + expect(JSON.parse(String(capture.init?.body))).toEqual({ + decision: "approve", + expected_version: "v3", + }); + expect(updated.status).toBe("promoted"); + }); + + it("surfaces a stale-version 409 as a proposal conflict", async () => { + respond(409, { + code: "proposal_conflict", + error: "proposal changed", + }); + const error = await decideLearningProposal("p1", "reject", "v1").catch( + (caught) => caught, + ); + expect(error).toBeInstanceOf(HarnessApiError); + expect(isProposalConflict(error)).toBe(true); + // A different code is NOT a conflict — flow control keys on the code. + expect( + isProposalConflict(new HarnessApiError(409, "dream_in_progress", "x")), + ).toBe(false); + // A code-less legacy daemon still classifies on the bare 409. + expect(isProposalConflict(new HarnessApiError(409, "", "x"))).toBe(true); + }); +}); + +describe("reflectHarnessSession", () => { + it("decodes the receipt counts", async () => { + const capture: { url?: string } = {}; + respond( + 200, + { + receipt: { + reflection_id: "r1", + disposition: "completed", + staged: 2, + promoted: 1, + conflicted: 0, + abstained: false, + }, + }, + capture, + ); + const receipt = await reflectHarnessSession("sess-1"); + expect(capture.url).toContain("/sessions/sess-1/reflect"); + expect(receipt).toEqual({ + reflectionId: "r1", + disposition: "completed", + queued: 0, + abstained: false, + staged: 2, + promoted: 1, + conflicted: 0, + }); + }); + + it("propagates a reflection failure as the typed error", async () => { + respond(500, { code: "internal", error: "explicit reflection failed" }); + await expect(reflectHarnessSession("sess-1")).rejects.toMatchObject({ + name: "HarnessApiError", + code: "internal", + status: 500, + }); + }); +}); diff --git a/studio/src/lib/harness/learning.ts b/studio/src/lib/harness/learning.ts new file mode 100644 index 0000000000..918e3904e8 --- /dev/null +++ b/studio/src/lib/harness/learning.ts @@ -0,0 +1,217 @@ +/** + * Learning review queue + explicit reflection (ADR 0109). + * + * Wire: `GET /v1/learning/proposals` (status/cursor/limit filters), + * `GET /v1/learning/proposals/{id}`, `POST .../decision` (approve/reject with + * `expected_version` — a stale version answers 409 `proposal_conflict`), + * `POST .../undo`, and `POST /v1/sessions/{id}/reflect`. + * + * Responses are stdlib JSON over the proto structs (snake_case keys, absent = + * zero value); gated by `capabilities.learning_proposals` / + * `capabilities.reflection` on GET /v1/compatibility. The `project` query is + * deliberately never sent: Studio reads the operator-scope partition — the + * browser never knows the workspace path (rule 2). + */ + +import { apiError, HARNESS_API, HarnessApiError } from "./client"; +import { + asArray, + asBool, + asNumber, + asRecord, + asString, + asStringArray, + timestampUnix, +} from "./wire"; + +/** One human decision recorded on a proposal. */ +interface LearningDecision { + kind: string; + actor: string; + reason: string; + atUnix: number; +} + +/** + * One learning proposal, decoded from the daemon's bounded digest projection. + * `status` is the daemon's vocabulary: staged (pending review), promoting, + * promoted, rejected, deferred_unsupported, conflicted, undone, + * skill_materialized. + */ +export interface LearningProposal { + id: string; + /** Optimistic-concurrency token — every decision/undo must carry it. */ + version: string; + status: string; + kind: string; + key: string; + value: string; + description: string; + title: string; + body: string; + triggers: string[]; + evidenceCount: number; + decisions: LearningDecision[]; + createdAtUnix: number; + updatedAtUnix: number; + projectScoped: boolean; + /** False when this partition has no trusted memory target — approve/undo disabled. */ + promotionAvailable: boolean; + promotionUnavailableReason: string; + /** Links a materialized procedure to its agent-owned learned skill. */ + learnedSkillId: string; +} + +export function decodeLearningProposal(raw: unknown): LearningProposal { + const record = asRecord(raw); + return { + id: asString(record.id), + version: asString(record.version), + status: asString(record.status), + kind: asString(record.kind), + key: asString(record.key), + value: asString(record.value), + description: asString(record.description), + title: asString(record.title), + body: asString(record.body), + triggers: asStringArray(record.triggers), + evidenceCount: asArray(record.evidence).length, + decisions: asArray(record.decisions).map((decision) => { + const entry = asRecord(decision); + return { + kind: asString(entry.kind), + actor: asString(entry.actor), + reason: asString(entry.reason), + atUnix: timestampUnix(entry.at), + }; + }), + createdAtUnix: timestampUnix(record.created_at), + updatedAtUnix: timestampUnix(record.updated_at), + projectScoped: asBool(record.project_scoped), + promotionAvailable: asBool(record.promotion_available), + promotionUnavailableReason: asString(record.promotion_unavailable_reason), + learnedSkillId: asString(record.learned_skill_id), + }; +} + +export interface LearningProposalPage { + proposals: LearningProposal[]; + nextCursor: string; +} + +export async function listLearningProposals( + options: { status?: string; cursor?: string; limit?: number } = {}, + signal?: AbortSignal, +): Promise { + const query = new URLSearchParams(); + if (options.status) query.set("status", options.status); + if (options.cursor) query.set("cursor", options.cursor); + if (options.limit) query.set("limit", String(options.limit)); + const suffix = query.size > 0 ? `?${query}` : ""; + const response = await fetch(`${HARNESS_API}/learning/proposals${suffix}`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + const body = asRecord(await response.json()); + return { + proposals: asArray(body.proposals).map(decodeLearningProposal), + nextCursor: asString(body.next_cursor), + }; +} + +/** + * Approves or rejects a proposal. `expectedVersion` is the version the review + * UI showed: a proposal that changed underneath answers 409 + * `proposal_conflict` (see isProposalConflict) — refresh and re-review, never + * blind-retry. + */ +export async function decideLearningProposal( + id: string, + decision: "approve" | "reject", + expectedVersion: string, + reason?: string, +): Promise { + const response = await fetch( + `${HARNESS_API}/learning/proposals/${encodeURIComponent(id)}/decision`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + decision, + expected_version: expectedVersion, + ...(reason ? { reason } : {}), + }), + }, + ); + if (!response.ok) throw await apiError(response); + return decodeLearningProposal(asRecord(await response.json()).proposal); +} + +/** Reverts a promoted proposal's memory write (same conflict contract). */ +export async function undoLearningPromotion( + id: string, + expectedVersion: string, +): Promise { + const response = await fetch( + `${HARNESS_API}/learning/proposals/${encodeURIComponent(id)}/undo`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ expected_version: expectedVersion }), + }, + ); + if (!response.ok) throw await apiError(response); + return decodeLearningProposal(asRecord(await response.json()).proposal); +} + +/** True when a decision/undo lost the optimistic-concurrency race. */ +export function isProposalConflict(error: unknown): boolean { + return ( + error instanceof HarnessApiError && + (error.code === "proposal_conflict" || + (error.code === "" && error.status === 409)) + ); +} + +/** The counts an explicit reflection pass returns. */ +export interface ReflectionReceipt { + reflectionId: string; + disposition: string; + queued: number; + abstained: boolean; + staged: number; + promoted: number; + conflicted: number; +} + +/** + * Runs an explicit reflection pass over one completed session + * (`POST /v1/sessions/{id}/reflect`). Synchronous: the request lasts the + * whole reflection run. Gated by `capabilities.reflection`. + */ +export async function reflectHarnessSession( + sessionId: string, + signal?: AbortSignal, +): Promise { + const response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/reflect`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: "{}", + signal, + }, + ); + if (!response.ok) throw await apiError(response); + const receipt = asRecord(asRecord(await response.json()).receipt); + return { + reflectionId: asString(receipt.reflection_id), + disposition: asString(receipt.disposition), + queued: asNumber(receipt.queued), + abstained: asBool(receipt.abstained), + staged: asNumber(receipt.staged), + promoted: asNumber(receipt.promoted), + conflicted: asNumber(receipt.conflicted), + }; +} diff --git a/studio/src/lib/harness/storage.test.ts b/studio/src/lib/harness/storage.test.ts new file mode 100644 index 0000000000..6420a4373a --- /dev/null +++ b/studio/src/lib/harness/storage.test.ts @@ -0,0 +1,89 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { + decodeStorageHealth, + fetchStorageHealth, + isStorageDegraded, +} from "./storage"; + +/** + * Pins the storage-health wire contract (ADR 0226): the stdlib-JSON subset the + * degraded-store banner reads, and the degraded classification. + */ + +const originalFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = originalFetch; +}); + +describe("decodeStorageHealth", () => { + it("decodes the banner subset and defaults absent fields", () => { + const health = decodeStorageHealth({ + available: true, + session_count: 12, + corrupt_count: 2, + v1_count: 3, + last_failure: "sweep: disk full", + }); + expect(health).toEqual({ + available: true, + unavailableReason: "", + sessionCount: 12, + corruptCount: 2, + v1Count: 3, + lastFailure: "sweep: disk full", + activeJob: "", + }); + }); + + it("never throws on a foreign shape", () => { + expect(decodeStorageHealth(null).available).toBe(false); + expect(decodeStorageHealth("x").corruptCount).toBe(0); + }); +}); + +describe("isStorageDegraded", () => { + const healthy = decodeStorageHealth({ available: true, session_count: 4 }); + it("healthy store → not degraded", () => { + expect(isStorageDegraded(healthy)).toBe(false); + }); + it("unavailable, corrupt families, or a recorded failure → degraded", () => { + expect(isStorageDegraded({ ...healthy, available: false })).toBe(true); + expect(isStorageDegraded({ ...healthy, corruptCount: 1 })).toBe(true); + expect(isStorageDegraded({ ...healthy, lastFailure: "boom" })).toBe(true); + }); + it("plain unmigrated v1 sessions still list, so they are not degraded", () => { + expect(isStorageDegraded({ ...healthy, v1Count: 9 })).toBe(false); + }); +}); + +describe("fetchStorageHealth", () => { + it("throws the typed error when the daemon refuses (management auth)", async () => { + globalThis.fetch = async () => + new Response( + JSON.stringify({ + code: "management_unauthorized", + error: "management authorization required", + }), + { + status: 403, + headers: { "Content-Type": "application/problem+json" }, + }, + ); + await expect(fetchStorageHealth()).rejects.toMatchObject({ + name: "HarnessApiError", + code: "management_unauthorized", + status: 403, + }); + }); + + it("decodes a healthy answer", async () => { + globalThis.fetch = async () => + new Response(JSON.stringify({ available: true, session_count: 2 }), { + status: 200, + headers: { "Content-Type": "application/json" }, + }); + const health = await fetchStorageHealth(); + expect(health.available).toBe(true); + expect(isStorageDegraded(health)).toBe(false); + }); +}); diff --git a/studio/src/lib/harness/storage.ts b/studio/src/lib/harness/storage.ts new file mode 100644 index 0000000000..e4d17184a8 --- /dev/null +++ b/studio/src/lib/harness/storage.ts @@ -0,0 +1,62 @@ +/** + * Storage health (ADR 0226): `GET /v1/storage/health`, gated by + * `capabilities.storage_health` on GET /v1/compatibility. + * + * Studio reads this for ONE purpose — the degraded-store banner that explains + * why sessions may be missing from the sidebar. Migration/cleanup job control + * stays CLI/TUI. The response is stdlib JSON over the proto struct + * (snake_case keys, absent = zero value); only the banner-relevant subset is + * decoded here. + */ + +import { apiError, HARNESS_API } from "./client"; +import { asBool, asNumber, asRecord, asString } from "./wire"; + +export interface StorageHealth { + /** False when the session store itself cannot be read. */ + available: boolean; + unavailableReason: string; + sessionCount: number; + /** Session families the store holds but cannot load — "missing" sessions. */ + corruptCount: number; + /** Legacy-layout families awaiting migration (CLI/TUI job). */ + v1Count: number; + /** The last background sweep/migration failure, "" when none. */ + lastFailure: string; + activeJob: string; +} + +export function decodeStorageHealth(raw: unknown): StorageHealth { + const record = asRecord(raw); + return { + available: asBool(record.available), + unavailableReason: asString(record.unavailable_reason), + sessionCount: asNumber(record.session_count), + corruptCount: asNumber(record.corrupt_count), + v1Count: asNumber(record.v1_count), + lastFailure: asString(record.last_failure), + activeJob: asString(record.active_job), + }; +} + +/** + * True when the store's state can explain sessions missing from the sidebar: + * the store is unreadable, some families no longer load, or a background job + * failed. Plain unmigrated v1 sessions still list, so they are NOT degraded. + */ +export function isStorageDegraded(health: StorageHealth): boolean { + return ( + !health.available || health.corruptCount > 0 || health.lastFailure !== "" + ); +} + +export async function fetchStorageHealth( + signal?: AbortSignal, +): Promise { + const response = await fetch(`${HARNESS_API}/storage/health`, { + signal, + cache: "no-store", + }); + if (!response.ok) throw await apiError(response); + return decodeStorageHealth(await response.json()); +} diff --git a/studio/src/lib/harness/watch.test.ts b/studio/src/lib/harness/watch.test.ts new file mode 100644 index 0000000000..ffc89f14e1 --- /dev/null +++ b/studio/src/lib/harness/watch.test.ts @@ -0,0 +1,196 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { HarnessApiError } from "./client"; +import { + parseSSEFrame, + type WatchDelivery, + WatchStreamError, + watchSessionEvents, +} from "./watch"; + +/** + * Pins the ADR-0250 watch client: the `event:`-aware SSE frame grammar (the + * prompt-stream parser drops tagged frames — this one must not), envelope + * delivery with cursor tracking, the typed terminal faults, and the + * reconnect-from-cursor path on a resumable `watch_lagging`. + */ + +describe("parseSSEFrame", () => { + it("reads the data line of a plain frame, with no event tag", () => { + expect(parseSSEFrame('data: {"phase":"live"}')).toEqual({ + event: "", + data: '{"phase":"live"}', + }); + }); + + it("reads an event:-tagged frame — the kind the prompt parser drops", () => { + expect( + parseSSEFrame('event: error\ndata: {"code":"watch_lagging"}'), + ).toEqual({ event: "error", data: '{"code":"watch_lagging"}' }); + }); + + it("joins multi-line data with newlines, per the EventSource grammar", () => { + expect(parseSSEFrame("data: line one\ndata: line two")).toEqual({ + event: "", + data: "line one\nline two", + }); + }); + + it("strips one leading space, tolerates CRLF, and skips comment lines", () => { + expect(parseSSEFrame(": keepalive\r\ndata: padded\r")).toEqual({ + event: "", + data: " padded", + }); + }); + + it("returns null for a frame with no data lines (a bare event tag)", () => { + expect(parseSSEFrame("event: error")).toBeNull(); + expect(parseSSEFrame(": comment only")).toBeNull(); + }); +}); + +// ── watchSessionEvents ─────────────────────────────────────────────────────── + +const originalFetch = globalThis.fetch; +afterEach(() => { + globalThis.fetch = originalFetch; +}); + +function sseResponse(frames: string[]): Response { + return new Response(frames.map((frame) => `${frame}\n\n`).join(""), { + status: 200, + headers: { "Content-Type": "text/event-stream" }, + }); +} + +const envelope = (body: unknown) => `data: ${JSON.stringify(body)}`; + +describe("watchSessionEvents", () => { + it("delivers replay frames, the live boundary, and tracks the cursor", async () => { + globalThis.fetch = vi.fn(async () => + sseResponse([ + envelope({ + event: { type: "user_prompt", user_prompt: { text: "hello" } }, + cursor: "c-1", + phase: "replay", + }), + envelope({ + event: { type: "message.delta", text: "hi", run_id: "run-1" }, + cursor: "c-2", + phase: "replay", + }), + // The silent lifecycle kind still advances the cursor (null event). + envelope({ + event: { type: "turn.end" }, + cursor: "c-3", + phase: "replay", + }), + envelope({ cursor: "c-3", phase: "live" }), + ]), + ); + const controller = new AbortController(); + const deliveries: WatchDelivery[] = []; + await watchSessionEvents( + "s-1", + (delivery) => { + deliveries.push(delivery); + // The boundary is the natural detach point for this test. + if (delivery.phase === "live" && !delivery.event) controller.abort(); + }, + { signal: controller.signal }, + ); + expect(deliveries).toEqual([ + { + phase: "replay", + cursor: "c-1", + event: { type: "user_prompt", text: "hello" }, + }, + { + phase: "replay", + cursor: "c-2", + event: { type: "token", text: "hi", runId: "run-1" }, + }, + { phase: "replay", cursor: "c-3", event: null }, + { phase: "live", cursor: "c-3", event: null }, + ]); + const url = (globalThis.fetch as ReturnType).mock + .calls[0][0] as string; + expect(url).toBe("/api/mecatl/v1/sessions/s-1/watch"); + }); + + it("reconnects from the last cursor on watch_lagging, bounded", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce( + sseResponse([ + envelope({ + event: { type: "message.delta", text: "a" }, + cursor: "c-7", + phase: "replay", + }), + 'event: error\ndata: {"code":"watch_lagging","error":"fell behind"}', + ]), + ) + .mockResolvedValueOnce( + sseResponse([ + 'event: error\ndata: {"code":"activity_gap","error":"append failed"}', + ]), + ); + globalThis.fetch = fetchMock; + const deliveries: WatchDelivery[] = []; + await expect( + watchSessionEvents("s-2", (delivery) => deliveries.push(delivery), { + runId: "run-9", + }), + ).rejects.toMatchObject({ + name: "WatchStreamError", + code: "activity_gap", + }); + expect(deliveries).toHaveLength(1); + expect(fetchMock).toHaveBeenCalledTimes(2); + // The reconnect resumes from the last processed cursor, same run filter. + const retryUrl = fetchMock.mock.calls[1][0] as string; + expect(retryUrl).toContain("cursor=c-7"); + expect(retryUrl).toContain("run_id=run-9"); + }); + + it("throws a typed WatchStreamError on cursor_expired — the transcript-refetch signal", async () => { + globalThis.fetch = vi.fn(async () => + sseResponse([ + 'event: error\ndata: {"code":"cursor_expired","error":"superseded"}', + ]), + ); + const fault = await watchSessionEvents("s-3", () => undefined).catch( + (caught) => caught, + ); + expect(fault).toBeInstanceOf(WatchStreamError); + expect((fault as WatchStreamError).code).toBe("cursor_expired"); + }); + + it("gives up after the reconnect budget when the stream keeps dropping", async () => { + const fetchMock = vi.fn(async () => sseResponse([])); + globalThis.fetch = fetchMock; + await expect( + watchSessionEvents("s-4", () => undefined, { maxReconnects: 2 }), + ).rejects.toBeInstanceOf(WatchStreamError); + // The first attempt plus two reconnects. + expect(fetchMock).toHaveBeenCalledTimes(3); + }); + + it("surfaces a pre-stream refusal as the typed HarnessApiError", async () => { + globalThis.fetch = vi.fn( + async () => + new Response( + JSON.stringify({ + code: "no_event_log", + error: "no durable event log configured", + }), + { status: 501 }, + ), + ); + const fault = await watchSessionEvents("s-5", () => undefined).catch( + (caught) => caught, + ); + expect(fault).toBeInstanceOf(HarnessApiError); + expect((fault as HarnessApiError).code).toBe("no_event_log"); + }); +}); diff --git a/studio/src/lib/harness/watch.ts b/studio/src/lib/harness/watch.ts new file mode 100644 index 0000000000..9874a2504a --- /dev/null +++ b/studio/src/lib/harness/watch.ts @@ -0,0 +1,276 @@ +/** + * Browser-side client for the durable session watch + * (`GET /v1/sessions/{id}/watch`, ADR 0250): SSE frames whose `data:` payload + * is a `{event, cursor, phase}` envelope — replay of what is already durable, + * one event-less boundary frame with phase "live", then live follow. + * + * Like client.ts this module owns TRANSPORT only; all wire decoding lives in + * the protocol seam. It has its own SSE parser because the watch route uses + * `event:`-tagged frames for terminal faults — the prompt stream's parser + * only ever looks at `data:` lines and would silently drop them, which is + * exactly the failure the tagged frame exists to abolish. + */ + +import type { StreamEvent } from "@/features/agent/types"; +import { parseWatchEnvelope, translateEvent } from "@/lib/protocol"; +import { apiError, HARNESS_API } from "./client"; + +/** One decoded SSE frame: the (optional) `event:` tag and the joined `data:` + * payload. Frames with no data lines carry nothing and decode to null. */ +export interface SSEFrame { + event: string; + data: string; +} + +/** + * Parses one SSE frame block (the text between blank-line separators) per the + * EventSource grammar: `field: value` lines split at the first colon, one + * optional leading space stripped from the value, multiple `data:` lines + * joined with newlines, comment lines (leading colon) ignored. + */ +export function parseSSEFrame(block: string): SSEFrame | null { + let event = ""; + const data: string[] = []; + for (const rawLine of block.split("\n")) { + const line = rawLine.endsWith("\r") ? rawLine.slice(0, -1) : rawLine; + if (!line || line.startsWith(":")) continue; + const colon = line.indexOf(":"); + const field = colon === -1 ? line : line.slice(0, colon); + let value = colon === -1 ? "" : line.slice(colon + 1); + if (value.startsWith(" ")) value = value.slice(1); + if (field === "event") event = value; + else if (field === "data") data.push(value); + } + if (data.length === 0) return null; + return { event, data: data.join("\n") }; +} + +/** + * A watch that ended on a stream fault, typed on the daemon's stable machine + * code (mid-stream faults ride an `event: error` frame — the 200 is already + * committed, so the status code is spent). The codes a caller branches on: + * - `watch_lagging`: resumable — the client fell behind; reconnect from the + * last cursor (this module already retried its bounded budget). + * - `activity_gap`: recorded events are missing; refetch the transcript. + * - `cursor_expired`: the cursor is from a superseded log generation; refetch + * the transcript and restart from the beginning. + */ +export class WatchStreamError extends Error { + readonly code: string; + constructor(code: string, message: string) { + super(message); + this.name = "WatchStreamError"; + this.code = code; + } +} + +/** One delivery to the watch consumer. */ +export interface WatchDelivery { + /** Open string: "replay" | "live" | "gap" — tolerate unknown values. */ + phase: string; + /** Opaque resume cursor positioned AFTER this envelope. */ + cursor: string; + /** Null on event-less frames — the replay→live boundary (phase "live") and + * gap markers (phase "gap") — and on envelopes whose event kinds have no + * visual surface (the cursor still advances). */ + event: StreamEvent | null; +} + +export interface WatchOptions { + /** Resume cursor; empty/absent = from the beginning (the normal first + * attachment). A cursor is scoped to the runId it was issued under. */ + cursor?: string; + /** Narrows delivery to one run's events (ADR 0249). */ + runId?: string; + signal?: AbortSignal; + /** Budget for RESUMABLE faults (watch_lagging, idle timeout, a dropped + * connection); any delivered frame refills it. */ + maxReconnects?: number; +} + +/** Mirrors client.ts's stream idle timeout: the watch route sends no + * keepalives, so a silent stretch (a parked approval can sit for hours) is + * handled by reconnecting from the last cursor, never by failing the UI. */ +const WATCH_IDLE_TIMEOUT_MS = 120_000; + +const DEFAULT_MAX_RECONNECTS = 5; + +/** Sentinel distinguishing the idle race from real read errors. */ +const idleTimeout = Symbol("watch-idle-timeout"); + +async function readOrIdle( + read: Promise, +): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(idleTimeout), WATCH_IDLE_TIMEOUT_MS); + }); + try { + return await Promise.race([read, timeout]); + } finally { + clearTimeout(timer); + } +} + +/** + * Attaches a durable watch and delivers every envelope to `onDelivery` until + * the caller aborts (resolves quietly) or the watch faults (throws). + * + * Resumable faults — `watch_lagging`, the idle timeout, a connection the + * daemon dropped without a terminal frame (e.g. a restart) — reconnect from + * the last processed cursor, bounded by `maxReconnects` per silent stretch + * (any delivered frame refills the budget). Non-resumable faults throw + * immediately: `activity_gap` / `cursor_expired` as WatchStreamError (the + * caller falls back to a transcript refetch), pre-stream refusals as + * HarnessApiError (`no_event_log` / `watch_unsupported` / 404 / 400). + */ +export async function watchSessionEvents( + sessionId: string, + onDelivery: (delivery: WatchDelivery) => void, + options?: WatchOptions, +): Promise { + const signal = options?.signal; + const maxReconnects = options?.maxReconnects ?? DEFAULT_MAX_RECONNECTS; + let cursor = options?.cursor ?? ""; + let reconnectsLeft = maxReconnects; + + // Each iteration is one connection attempt; `continue` reconnects from the + // last processed cursor after a resumable fault. + while (true) { + if (signal?.aborted) return; + const query = new URLSearchParams(); + if (cursor) query.set("cursor", cursor); + if (options?.runId) query.set("run_id", options.runId); + const queryString = query.toString(); + const suffix = queryString ? `?${queryString}` : ""; + let response: Response; + try { + response = await fetch( + `${HARNESS_API}/sessions/${encodeURIComponent(sessionId)}/watch${suffix}`, + { signal, cache: "no-store" }, + ); + } catch (caught) { + // A caller-initiated abort resolves quietly; a real network fault + // (daemon offline) propagates — the connection gate owns that state. + if (signal?.aborted) return; + throw caught; + } + // Pre-stream refusals are real HTTP statuses with a problem body: a + // missing feature (501 no_event_log / watch_unsupported), an unknown + // session (404), a delegation-child id (400). Never retried here. + if (!response.ok) throw await apiError(response); + if (!response.body) throw new Error("harness returned no event stream"); + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let stalled = false; + try { + while (true) { + const result = await readOrIdle(reader.read()); + if (result === idleTimeout) { + // No keepalives on this route: silence is ambiguous between a + // genuinely quiet session and a dead socket. Reconnecting from the + // cursor is correct in both cases (at-least-once delivery). + stalled = true; + break; + } + if (result.done) { + // The daemon never ends a healthy watch — a clean close without a + // terminal frame is a restart or a dropped proxy. Resumable. + stalled = true; + break; + } + buffer += decoder.decode(result.value, { stream: true }); + const blocks = buffer.split("\n\n"); + buffer = blocks.pop() ?? ""; + for (const block of blocks) { + const frame = parseSSEFrame(block); + if (!frame) continue; + if (frame.event === "error") { + // Terminal fault after the 200: {code, error} on the data line. + let code = ""; + let detail = frame.data; + try { + const body = JSON.parse(frame.data) as { + code?: string; + error?: string; + }; + code = typeof body.code === "string" ? body.code : ""; + detail = body.error ?? frame.data; + } catch { + // A malformed fault frame still terminates; code stays "". + } + if (code === "watch_lagging") { + // Resumable by contract: reconnect from the last processed + // cursor. The shared budget check at the bottom bounds it. + stalled = true; + break; + } + throw new WatchStreamError(code, detail); + } + let delivered: WatchDelivery[]; + try { + const envelope = parseWatchEnvelope(frame.data); + const events = envelope.event + ? translateEvent(envelope.event, sessionId) + : []; + delivered = + events.length > 0 + ? events.map((event) => ({ + phase: envelope.phase, + cursor: envelope.cursor, + event, + })) + : [ + // Event-less frames (the live boundary, gap markers) and + // envelopes whose kinds render nothing still advance the + // caller's cursor and carry the phase. + { + phase: envelope.phase, + cursor: envelope.cursor, + event: null, + }, + ]; + } catch { + // A frame this Studio version cannot decode is surfaced to the + // caller as an unrenderable notice, never silently dropped. + delivered = [ + { + phase: "", + cursor, + event: { + type: "notice", + text: "Mecatl sent a watch frame this Studio version could not decode.", + }, + }, + ]; + } + for (const delivery of delivered) { + if (delivery.cursor) cursor = delivery.cursor; + onDelivery(delivery); + } + // Progress proves the wire is healthy; refill the retry budget. + reconnectsLeft = maxReconnects; + } + if (stalled) break; + } + } catch (caught) { + if (signal?.aborted) return; + if (caught instanceof WatchStreamError) throw caught; + // A network read error mid-stream is resumable, like a dropped socket. + stalled = true; + } finally { + reader.cancel().catch(() => undefined); + } + if (signal?.aborted) return; + if (!stalled) return; + if (reconnectsLeft <= 0) { + throw new WatchStreamError( + "watch_lagging", + "The watch kept stalling and its reconnect budget is spent.", + ); + } + reconnectsLeft -= 1; + } +} diff --git a/studio/src/lib/harness/wire.ts b/studio/src/lib/harness/wire.ts new file mode 100644 index 0000000000..33f1b08a1c --- /dev/null +++ b/studio/src/lib/harness/wire.ts @@ -0,0 +1,50 @@ +/** + * Tolerant field readers shared by the daemon wire modules that decode + * stdlib-JSON-encoded proto messages (learning, learned skills, dream, + * storage health). + * + * The daemon marshals these responses with Go's encoding/json over the + * generated proto structs, so keys are snake_case proto field names, absent + * means zero-value (omitempty), and google.protobuf.Timestamp arrives as the + * struct `{seconds, nanos}` — NOT the protojson RFC 3339 string. Every reader + * here degrades to a zero value rather than throwing, so a shape drift renders + * as missing data, never a broken page. + */ + +export function asString(raw: unknown): string { + return typeof raw === "string" ? raw : ""; +} + +export function asNumber(raw: unknown): number { + if (typeof raw === "number" && Number.isFinite(raw)) return raw; + if (typeof raw === "string") { + const parsed = Number(raw); + return Number.isFinite(parsed) ? parsed : 0; + } + return 0; +} + +export function asBool(raw: unknown): boolean { + return raw === true; +} + +export function asRecord(raw: unknown): Record { + return typeof raw === "object" && raw !== null && !Array.isArray(raw) + ? (raw as Record) + : {}; +} + +export function asArray(raw: unknown): unknown[] { + return Array.isArray(raw) ? raw : []; +} + +export function asStringArray(raw: unknown): string[] { + return asArray(raw).filter( + (item): item is string => typeof item === "string", + ); +} + +/** Unix seconds off a stdlib-JSON proto Timestamp (`{seconds, nanos}`). */ +export function timestampUnix(raw: unknown): number { + return asNumber(asRecord(raw).seconds); +} diff --git a/studio/src/lib/protocol/index.ts b/studio/src/lib/protocol/index.ts new file mode 100644 index 0000000000..f4af950948 --- /dev/null +++ b/studio/src/lib/protocol/index.ts @@ -0,0 +1,24 @@ +export { + parseMecatlEvent, + parseWatchEnvelope, + translateEvent, +} from "./events"; +export { + decodeScheduleFires, + decodeScheduleRows, + encodeScheduleSpec, + type ScheduleCarriedSpec, + type ScheduleFireRow, + type ScheduleRow, + type ScheduleSpecDraft, +} from "./schedules"; +export { + decodeSessionInventory, + decodeSessionPermissionMode, + decodeSessionTranscript, + encodeSessionPermissionMode, + type SessionInventoryPage, + type SessionPermissionMode, + type SessionSummary, + type SessionTranscript, +} from "./sessions"; diff --git a/studio/src/lib/protocol/sessions.ts b/studio/src/lib/protocol/sessions.ts index 70495fa1cd..e239d0fc72 100644 --- a/studio/src/lib/protocol/sessions.ts +++ b/studio/src/lib/protocol/sessions.ts @@ -16,7 +16,7 @@ import { optionalString, } from "./internal"; -type SessionSummary = { +export type SessionSummary = { sessionId: string; /** The server-held label: operator-authored, or seeded from the first prompt. */ title: string;