/**
 * LoomcycleClient — the single public class exported by
 * @loomcycle/client. Speaks HTTP+SSE to a running loomcycle sidecar.
 *
 * hooks-connector PR C: full Python-adapter parity + hook management.
 * 27 methods total — 26 async (run streaming, continuation, agent
 * metadata, transcript, health, users, pause/resume/state, snapshot
 * lifecycle capture / list / get / restore / delete, memory admin,
 * interruption listing + resolve, hook registration / list / delete)
 * plus one synchronous helper (exportSnapshotURL builds a URL string
 * without issuing a request).
 *
 * Construction:
 *
 *   const client = new LoomcycleClient({
 *     baseUrl: "http://127.0.0.1:8787",  // or process.env.LOOMCYCLE_BASE_URL
 *     authToken: "...",                  // or process.env.LOOMCYCLE_AUTH_TOKEN
 *   });
 *
 * Streaming methods (`runStreaming`, `continueSession`) return
 * AsyncIterable<AgentEvent>; non-streaming methods return
 * Promise<T>. Non-2xx responses throw typed errors from errors.ts
 * via fetch-helpers.ts:raiseFromResponse — see README.md for the
 * full mapping table.
 */

import type { _FetchContext } from "./fetch-helpers.js";
import {
  authHeaders,
  deleteRequest,
  jsonFetch,
  patchJSON,
  postJSON,
  putJSON,
  raiseFromResponse,
} from "./fetch-helpers.js";
import { parseSSE } from "./stream.js";
import { InteractiveSession } from "./interactive.js";
import {
  ClientToolHost,
  clientToolsURL,
  type ConnectClientToolsOptions,
} from "./client-tools.js";
import type {
  DirectoryInspection,
  DirectoryTenant,
  DirectoryUser,
  ErasureReport,
  ErasureResult,
  Agent,
  AgentEvent,
  AgentStatus,
  CancelAgentResult,
  ClientOptions,
  AgentDefOverlay,
  AgentDefRowResponse,
  ContinueOptions,
  CreateSnapshotOptions,
  EnsureCodeAgentOptions,
  EnsureCodeAgentResult,
  EnsureMcpServerOptions,
  EnsureMcpServerResult,
  HealthResponse,
  Hook,
  InterruptListResponse,
  InterruptStatus,
  ListAgentsResponse,
  AckChannelOptions,
  AwaitChannelsOptions,
  BroadcastChannelsOptions,
  ChannelAckResult,
  ChannelAwaitResult,
  ChannelBroadcastResult,
  ChannelDescriptor,
  ChannelPeekResult,
  ChannelPublishResult,
  ChannelPurgeResult,
  ChannelSubscribeResult,
  CreateChannelOptions,
  ListChannelsResponse,
  PeekChannelOptions,
  PublishChannelOptions,
  SetMemoryEntryOptions,
  SetMemoryEntryResponse,
  SubscribeChannelOptions,
  UpdateChannelOptions,
  LibraryAgentDefinition,
  LibraryListResponse,
  LibraryMcpServerDefinition,
  LibrarySkillDefinition,
  ListHooksResponse,
  ListUsersResponse,
  LLMChatOptions,
  LLMChatResponse,
  LLMChatStreamItem,
  LLMEmbeddingsOptions,
  LLMEmbeddingsResponse,
  MCPServerDefRowResponse,
  MCPServerDefVerifyResult,
  MemoryEmbedStatsResponse,
  MemoryEntriesResponse,
  MemoryEntryResponse,
  MemoryReembedResponse,
  MemoryScopeIDsResponse,
  MemoryScopesResponse,
  MemorySearchInput,
  MemorySearchResponse,
  PauseResult,
  PersistentVolumesResponse,
  EphemeralVolumesResponse,
  RegisterHookOptions,
  RegisterHookResponse,
  ResolveInterruptOptions,
  ResumeResult,
  ResolverMatrix,
  CompactRunResult,
  ReplaySessionResult,
  CancelTurnResult,
  CompactionOptions,
  RunBatchOptions,
  RunBatchResult,
  RunOptions,
  RunStateStreamItem,
  SamplingOptions,
  RuntimeStateResponse,
  SnapshotCreateResponse,
  SnapshotDescriptor,
  SnapshotEnvelope,
  SnapshotListResponse,
  SnapshotRestoreResponse,
  StreamUserRunStatesOptions,
  SubstrateToolInput,
  SubstrateToolResponse,
  CreatedTeam,
  ListTeamsResponse,
  TeamDefDetail,
  TeamDiagram,
  TeamRunResult,
  PathToolInput,
  PathToolResponse,
  DocumentToolInput,
  DocumentToolResponse,
  HistoryToolInput,
  HistoryToolResponse,
  TranscriptResponse,
  WhoamiResponse,
  UsageDimension,
  UsageReportResponse,
  TokenLimit,
  TokenLimitsResponse,
  ConfigResponse,
  SetTokenLimitRequest,
} from "./types.js";

/** samplingToWire maps the camelCase SamplingOptions to the snake_case wire
 *  object, omitting unset fields so they inherit the agent's value. */
function samplingToWire(s: SamplingOptions): Record<string, unknown> {
  const w: Record<string, unknown> = {};
  if (s.temperature !== undefined) w.temperature = s.temperature;
  if (s.topP !== undefined) w.top_p = s.topP;
  if (s.topK !== undefined) w.top_k = s.topK;
  if (s.frequencyPenalty !== undefined) w.frequency_penalty = s.frequencyPenalty;
  if (s.presencePenalty !== undefined) w.presence_penalty = s.presencePenalty;
  if (s.seed !== undefined) w.seed = s.seed;
  if (s.stop !== undefined) w.stop = s.stop;
  return w;
}

/** compactionToWire maps the camelCase CompactionOptions to the snake_case
 *  wire object, omitting unset fields so they inherit the agent's value. */
function compactionToWire(c: CompactionOptions): Record<string, unknown> {
  const w: Record<string, unknown> = {};
  if (c.enabled !== undefined) w.enabled = c.enabled;
  if (c.targetPercentage !== undefined) w.target_percentage = c.targetPercentage;
  if (c.keepLastN !== undefined) w.keep_last_n = c.keepLastN;
  if (c.keepFirst !== undefined) w.keep_first = c.keepFirst;
  if (c.autocompactAtPct !== undefined) w.autocompact_at_pct = c.autocompactAtPct;
  if (c.model !== undefined) w.model = c.model;
  return w;
}

/** runBody builds the snake_case /v1/runs request body from RunOptions,
 *  omitting unset fields (preserves the server's nil semantics — notably
 *  `allowedHosts: null` is treated as "omit", not deny-all). Shared by
 *  runStreaming and each spawn of spawnRunBatch. `signal`/`debug` are client
 *  concerns and never sent. */
function runBody(opts: RunOptions): Record<string, unknown> {
  const body: Record<string, unknown> = {
    agent: opts.agent,
    segments: opts.segments,
  };
  if (opts.tools !== undefined) body.tools = opts.tools;
  if (opts.allowedHosts !== undefined && opts.allowedHosts !== null) {
    body.allowed_hosts = opts.allowedHosts;
  }
  if (opts.webSearchFilter !== undefined) body.web_search_filter = opts.webSearchFilter;
  if (opts.sessionId !== undefined) body.session_id = opts.sessionId;
  if (opts.tenantId !== undefined) body.tenant_id = opts.tenantId;
  if (opts.userId !== undefined) body.user_id = opts.userId;
  if (opts.agentId !== undefined) body.agent_id = opts.agentId;
  if (opts.userTier !== undefined) body.user_tier = opts.userTier;
  if (opts.userBearer !== undefined) body.user_bearer = opts.userBearer;
  if (opts.userCredentials !== undefined) body.user_credentials = opts.userCredentials;
  if (opts.parentContext !== undefined) body.parent_context = opts.parentContext;
  if (opts.metadata !== undefined) body.metadata = opts.metadata;
  if (opts.runTimeoutSeconds !== undefined) body.run_timeout_seconds = opts.runTimeoutSeconds;
  if (opts.sampling !== undefined) body.sampling = samplingToWire(opts.sampling);
  if (opts.compaction !== undefined) body.compaction = compactionToWire(opts.compaction);
  if (opts.interactive !== undefined) body.interactive = opts.interactive;
  return body;
}

export class LoomcycleClient {
  private ctx: _FetchContext;

  constructor(opts: ClientOptions = {}) {
    this.ctx = {
      baseUrl: (opts.baseUrl ?? "http://127.0.0.1:8787").replace(/\/$/, ""),
      authToken: opts.authToken,
      fetchImpl: opts.fetch ?? fetch,
    };
  }

  // ---- Run lifecycle ----

  /**
   * Run an agent and stream events. Returns AsyncIterable<AgentEvent>;
   * the iterator completes when the server closes the SSE stream.
   *
   * Errors during the run surface as `{ type: "error", error }` events;
   * only transport / HTTP-level failures throw — and those throw typed
   * errors (e.g. AuthError for 401, BackpressureError for 429).
   *
   * **Blocking semantics.** This iterator is alive for the FULL
   * duration of the run — typically seconds, occasionally minutes for
   * long tool chains. Callers that need fire-and-forget completion
   * notifications (n8n's worker model, dashboards that don't want to
   * hold a connection per active run) should subscribe to
   * {@link LoomcycleClient.streamUserRunStates} instead, which yields
   * one terminal-state frame per completed run without holding the
   * run's stream open.
   *
   * v0.9.x — pass `opts.debug = true` to emit synthetic
   * `{ type: "_meta", meta_subtype: "stream_open" | "stream_close" }`
   * events around the real frames. Silent (default) when omitted.
   */
  async *runStreaming(opts: RunOptions): AsyncIterable<AgentEvent> {
    yield* this.streamSSE("/v1/runs", runBody(opts), opts.signal, opts.debug);
  }

  /**
   * Continue an existing session with a new run. The session's prior
   * transcript is replayed into the model's context server-side;
   * this iterator yields only the NEW events from the continuation.
   *
   * Raises SessionNotFoundError when sessionId is unknown,
   * SessionBusyError when another request is in flight on the same
   * session.
   *
   * **Blocking semantics.** Same as {@link LoomcycleClient.runStreaming} —
   * the iterator stays alive for the duration of the new run. For
   * async fire-and-forget completion patterns, see
   * {@link LoomcycleClient.streamUserRunStates}.
   *
   * v0.9.x — pass `opts.debug = true` for synthetic
   * `_meta` open/close events.
   */
  async *continueSession(opts: ContinueOptions): AsyncIterable<AgentEvent> {
    const body: Record<string, unknown> = {
      segments: opts.segments,
    };
    if (opts.tools !== undefined) body.tools = opts.tools;
    if (opts.allowedHosts !== undefined && opts.allowedHosts !== null) {
      body.allowed_hosts = opts.allowedHosts;
    }
    if (opts.webSearchFilter !== undefined) body.web_search_filter = opts.webSearchFilter;
    if (opts.agentId !== undefined) body.agent_id = opts.agentId;
    if (opts.userTier !== undefined) body.user_tier = opts.userTier;
    if (opts.userBearer !== undefined) body.user_bearer = opts.userBearer;
    if (opts.userCredentials !== undefined) body.user_credentials = opts.userCredentials;
    if (opts.parentContext !== undefined) body.parent_context = opts.parentContext;
    if (opts.metadata !== undefined) body.metadata = opts.metadata;
    if (opts.runTimeoutSeconds !== undefined) body.run_timeout_seconds = opts.runTimeoutSeconds;
    if (opts.sampling !== undefined) body.sampling = samplingToWire(opts.sampling);
    if (opts.compaction !== undefined) body.compaction = compactionToWire(opts.compaction);
    if (opts.interactive !== undefined) body.interactive = opts.interactive;
    yield* this.streamSSE(
      `/v1/sessions/${encodeURIComponent(opts.sessionId)}/messages`,
      body,
      opts.signal,
      opts.debug,
    );
  }

  /** Push an operator steering message into a LIVE interactive run (RFC AI).
   *  Mirrors `POST /v1/runs/{run_id}/input`. The run must be in-flight —
   *  parked at end_turn awaiting input, or mid-turn (the message is drained
   *  at the next iteration boundary). Does NOT open a stream; the operator's
   *  turn + the model's response arrive on the run's OWN event stream
   *  (the open `runStreaming` iterator, or a `streamRunByID` re-attach).
   *  Returns `{ run_id, delivered }`.
   *
   *  Raises {@link UnavailableError} (503, steering off / no run),
   *  {@link AuthError} (401). A full steer queue surfaces as a 429. */
  async sendRunInput(
    runId: string,
    text: string,
    opts?: { signal?: AbortSignal },
  ): Promise<{ run_id: string; delivered: boolean }> {
    return postJSON<{ run_id: string; delivered: boolean }>(
      this.ctx,
      `/v1/runs/${encodeURIComponent(runId)}/input`,
      { text },
      opts,
    );
  }

  /** Re-attach to a run's event stream by `run_id` (RFC AI), replaying from
   *  `fromSeq` (default 0) then live-tailing — the gRPC-free twin of the Web
   *  UI's resume-in-terminal. Mirrors `GET /v1/runs/{run_id}/stream`. The
   *  operator's own prior turns are replayed as `steer` events
   *  (`user_input.source === "replay"`), so a cold client — e.g. resuming on
   *  another device — reconstructs the whole conversation. A PARKED
   *  interactive run keeps streaming until it ends or `signal` aborts; a
   *  disconnect does NOT stop the run.
   *
   *  The first frame is an `agent` side-channel carrying the run's
   *  agent_id / session_id (for `sendRunInput` / `cancelAgent`). */
  async *streamRunByID(
    runId: string,
    opts?: { fromSeq?: number; signal?: AbortSignal },
  ): AsyncIterable<AgentEvent> {
    const q = opts?.fromSeq ? `?from_seq=${opts.fromSeq}` : "";
    yield* this.streamSSEGet(
      `/v1/runs/${encodeURIComponent(runId)}/stream${q}`,
      opts?.signal,
    );
  }

  /** Start a high-level INTERACTIVE session (RFC AI) — the ergonomic driver
   *  for a persistent run you converse with turn by turn. Starts an
   *  `interactive` run and returns an {@link InteractiveSession} whose
   *  `events()` you iterate and whose `send()` / `cancel()` steer it. See
   *  {@link InteractiveSession}. For the raw primitives, use
   *  {@link LoomcycleClient.runStreaming} with `interactive: true` +
   *  {@link LoomcycleClient.sendRunInput}. */
  interactiveSession(opts: Omit<RunOptions, "interactive">): InteractiveSession {
    const source = this.runStreaming({ ...opts, interactive: true });
    return new InteractiveSession(source, {
      sendRunInput: (rid, t) => this.sendRunInput(rid, t),
      cancelAgent: (aid) => this.cancelAgent(aid),
    });
  }

  /** Re-attach to a detached interactive run by `run_id` as a high-level
   *  {@link InteractiveSession} — e.g. resuming on another device. The run_id
   *  is known up front, so `send()` works immediately. Replays the
   *  conversation (incl. the operator's prior turns, RFC AI) then live-tails. */
  attachInteractiveSession(
    runId: string,
    opts?: { fromSeq?: number; signal?: AbortSignal },
  ): InteractiveSession {
    const source = this.streamRunByID(runId, opts);
    const session = new InteractiveSession(source, {
      sendRunInput: (rid, t) => this.sendRunInput(rid, t),
      cancelAgent: (aid) => this.cancelAgent(aid),
    });
    session.runId = runId;
    return session;
  }

  /**
   * Spawn N fresh runs concurrently in ONE call (RFC Y external fan-out) and
   * resolve once they ALL settle, returning the combined index-aligned
   * envelope. A per-child failure is captured in that child's result
   * (`status` + `error`) and never rejects the batch. Prefer this over firing
   * N parallel {@link LoomcycleClient.runStreaming} calls.
   *
   * Each spawn is a fresh run (its `sessionId` is ignored). Capped at 32 —
   * an over-cap batch rejects with InvalidArgumentError. `mode: "detach"`
   * (async handles) is reserved for a future release and rejected today.
   *
   * Blocking: resolves only when the slowest child finishes (or `timeoutMs`
   * elapses). Mirrors POST /v1/runs:batch.
   */
  async spawnRunBatch(opts: RunBatchOptions): Promise<RunBatchResult> {
    const body: Record<string, unknown> = {
      spawns: opts.spawns.map(runBody),
    };
    if (opts.mode !== undefined) body.mode = opts.mode;
    if (opts.timeoutMs !== undefined) body.timeout_ms = opts.timeoutMs;
    return postJSON<RunBatchResult>(this.ctx, "/v1/runs:batch", body, {
      signal: opts.signal,
    });
  }

  /**
   * Compact a run's conversation: summarize the history to free context and
   * continue from the summary. Targets the run by `runId`. A live run must be
   * PARKED (awaiting input) — a mid-turn run rejects (RunBusyError / 409).
   * Returns `{ compacted, before_tokens, after_tokens, applied }`, where
   * `applied` is "live", "marker", or "noop". Mirrors
   * POST /v1/runs/{run_id}/compact.
   */
  /**
   * List the subjects with activity in a tenant (loomcycle v1.46.0+).
   *
   * A "user" here is DERIVED from run activity, not a stored record — there is no
   * create or update, and removing a subject's footprint is
   * {@link erasureExecute}.
   */
  async directoryUsers(
    opts?: { tenant?: string; signal?: AbortSignal },
  ): Promise<{ users: DirectoryUser[] }> {
    const q = opts?.tenant !== undefined
      ? `?tenant=${encodeURIComponent(opts.tenant)}`
      : "";
    return jsonFetch<{ users: DirectoryUser[] }>(this.ctx, `/v1/_users${q}`, opts);
  }

  /**
   * Aggregate one subject's activity, chats, memory, documents, budget and usage
   * in a single call (loomcycle v1.46.0+).
   *
   * An admin token MUST pass `tenant` — a subject id is only unique within one, so
   * the server refuses rather than guessing (`""` selects the default tenant on a
   * single-tenant install). A tenant-scoped principal is confined regardless.
   */
  async directoryInspect(
    subject: string,
    opts?: { tenant?: string; signal?: AbortSignal },
  ): Promise<DirectoryInspection> {
    const q = opts?.tenant !== undefined
      ? `?tenant=${encodeURIComponent(opts.tenant)}`
      : "";
    return jsonFetch<DirectoryInspection>(
      this.ctx,
      `/v1/_users/${encodeURIComponent(subject)}${q}`,
      opts,
    );
  }

  /**
   * Enumerate tenants with derived counts (loomcycle v1.46.0+). **Requires an
   * operator-admin token** — the list of tenants is itself cross-tenant
   * information, so a scoped token is refused rather than given a filtered list.
   *
   * Derived from runs: a tenant that has never started a run does not appear, so
   * an empty list means no ACTIVITY, not no tenants.
   */
  async directoryTenants(
    opts?: { signal?: AbortSignal },
  ): Promise<{ tenants: DirectoryTenant[]; notes?: string[] }> {
    return jsonFetch<{ tenants: DirectoryTenant[]; notes?: string[] }>(
      this.ctx,
      "/v1/_tenants",
      opts,
    );
  }

  /**
   * Report what this deployment holds about one SUBJECT, in three tiers
   * separated by what they guarantee (RFC BL P5). Read-only.
   *
   * `tenant` is optional and honored only for an admin token; a tenant-scoped
   * principal is confined to its own tenant regardless. An admin MUST pass it —
   * a subject id is only unique within one tenant, so the server refuses rather
   * than guessing (pass `""` for the default tenant on a single-tenant install).
   */
  async erasureReport(
    subject: string,
    opts?: { tenant?: string; signal?: AbortSignal },
  ): Promise<ErasureReport> {
    const q = new URLSearchParams({ subject });
    if (opts?.tenant !== undefined) q.set("tenant", opts.tenant);
    return jsonFetch<ErasureReport>(this.ctx, `/v1/_erasure?${q}`, opts);
  }

  /**
   * Erase tiers 1 and 2 for a SUBJECT, reporting what could not be reached.
   *
   * DEFAULTS TO A DRY RUN. `dryRun` defaults to true and a live run additionally
   * requires `confirm` to equal `subject`; omitting either deletes nothing.
   *
   * IMPORTANT: tier-3 residue is traceable only through the subject's chats,
   * which a live run deletes — so a report afterwards shows `rows: 0` while
   * those facts remain. The returned object is the ONLY durable record of what
   * was not reached; persist it.
   */
  async erasureExecute(
    subject: string,
    opts?: {
      dryRun?: boolean;
      confirm?: string;
      tenant?: string;
      signal?: AbortSignal;
    },
  ): Promise<ErasureResult> {
    const body: Record<string, unknown> = { subject };
    // Sent explicitly rather than relying on the server default, so a caller
    // reading this code sees which mode the request is in.
    body.dry_run = opts?.dryRun ?? true;
    if (opts?.confirm !== undefined) body.confirm = opts.confirm;
    const q = opts?.tenant !== undefined
      ? `?tenant=${encodeURIComponent(opts.tenant)}`
      : "";
    return postJSON<ErasureResult>(this.ctx, `/v1/_erasure${q}`, body, opts);
  }

  async compactRun(
    runId: string,
    opts?: { reason?: string; signal?: AbortSignal },
  ): Promise<CompactRunResult> {
    const body: Record<string, unknown> = {};
    if (opts?.reason !== undefined) body.reason = opts.reason;
    return postJSON<CompactRunResult>(
      this.ctx,
      `/v1/runs/${encodeURIComponent(runId)}/compact`,
      body,
      opts,
    );
  }

  /**
   * Replay a session's transcript into a NEW session bound to a (possibly
   * different) target agent, so that agent continues from the same context
   * (RFC BJ Phase 4). `sourceSessionId` is the source; the returned
   * `new_session_id` is a fresh session you continue with the normal message
   * path — the carried context replays automatically. Pass `compress` to
   * collapse the carried history to a summary + recent tail.
   */
  async replaySession(
    sourceSessionId: string,
    opts: { agent: string; compress?: boolean; signal?: AbortSignal },
  ): Promise<ReplaySessionResult> {
    const body: Record<string, unknown> = { agent: opts.agent };
    if (opts.compress !== undefined) body.compress = opts.compress;
    return postJSON<ReplaySessionResult>(
      this.ctx,
      `/v1/sessions/${encodeURIComponent(sourceSessionId)}/replay`,
      body,
      opts,
    );
  }

  /**
   * Stop the CURRENT turn of a live interactive run (its in-flight generation +
   * the tool calls it started) and park it at awaiting_input — session +
   * transcript intact, ready for the next message. This is the "Esc" gesture,
   * NOT whole-run cancel ({@link LoomcycleClient.cancelAgent}), which terminates
   * the run. Returns `{ run_id, stopped, parked }`.
   *
   * Rejects with RunBusyError (409) when the run isn't mid-turn or isn't
   * interactive, and AgentNotFoundError (404) for an unknown / cross-tenant run.
   * Mirrors POST /v1/runs/{run_id}/cancel. (RFC BH)
   */
  async cancelTurn(
    runId: string,
    opts?: { reason?: string; signal?: AbortSignal },
  ): Promise<CancelTurnResult> {
    const body: Record<string, unknown> = {};
    if (opts?.reason !== undefined) body.reason = opts.reason;
    return postJSON<CancelTurnResult>(
      this.ctx,
      `/v1/runs/${encodeURIComponent(runId)}/cancel`,
      body,
      opts,
    );
  }

  // ---- Agent metadata ----

  /** Read one agent's status + usage stats. Raises AgentNotFoundError
   *  when the agent_id is unknown. */
  async getAgent(agentId: string, opts?: { signal?: AbortSignal }): Promise<Agent> {
    return jsonFetch<Agent>(this.ctx, `/v1/agents/${encodeURIComponent(agentId)}`, opts);
  }

  /** Cancel a live agent (cascades to children via parent_agent_id).
   *  Returns count of agents cancelled. Idempotent — already-terminated
   *  agents return 0. */
  async cancelAgent(
    agentId: string,
    opts?: { reason?: string; signal?: AbortSignal },
  ): Promise<CancelAgentResult> {
    const resp = await postJSON<{ cancelled_count: number }>(
      this.ctx,
      `/v1/agents/${encodeURIComponent(agentId)}/cancel`,
      { reason: opts?.reason ?? "" },
      opts,
    );
    return { cancelledCount: resp.cancelled_count };
  }

  /** List a user's recent agent runs, optionally filtered by status.
   *
   *  v0.9.x — `parentAgentId` narrows the result CLIENT-SIDE to runs
   *  whose `parent_agent_id` matches. The server still returns the
   *  full set (server-side `?parent_agent_id=` filter is a future
   *  request); the adapter trims before returning. Useful for the
   *  n8n trigger pattern "show me all sub-runs spawned by parent X." */
  async listUserAgents(
    userId: string,
    opts?: {
      status?: AgentStatus;
      parentAgentId?: string;
      /** Super-admin tenant-focus (?tenant=, RFC L v0.17.0). Ignored
       *  server-side for a tenant principal — its own tenant is forced. */
      tenant?: string;
      signal?: AbortSignal;
    },
  ): Promise<Agent[]> {
    const params = new URLSearchParams();
    if (opts?.status) params.set("status", opts.status);
    if (opts?.tenant) params.set("tenant", opts.tenant);
    const q = params.toString() ? `?${params.toString()}` : "";
    const resp = await jsonFetch<ListAgentsResponse>(
      this.ctx,
      `/v1/users/${encodeURIComponent(userId)}/agents${q}`,
      opts,
    );
    const all = resp.agents ?? [];
    if (opts?.parentAgentId !== undefined && opts.parentAgentId !== "") {
      return all.filter((a) => a.parent_agent_id === opts.parentAgentId);
    }
    return all;
  }

  /** Aggregated token-usage + cost report (RFC AV). Group by any of
   *  tenant/user/provider/model/source over an optional time window; group by
   *  `source` for the operator-vs-tenant split. Tenant-scoped server-side: a
   *  tenant principal sees only its own tenant (the `tenant` focus is
   *  admin-only). Mirrors `GET /v1/_usage`. */
  async usageReport(opts?: {
    /** Dimensions to group by (default server-side: tenant,source). */
    groupBy?: UsageDimension[];
    /** RFC3339 window bounds (inclusive); omit for unbounded. */
    from?: string;
    to?: string;
    /** Super-admin tenant focus (?tenant=); ignored for a tenant principal. */
    tenant?: string;
    signal?: AbortSignal;
  }): Promise<UsageReportResponse> {
    const params = new URLSearchParams();
    if (opts?.groupBy && opts.groupBy.length > 0)
      params.set("group_by", opts.groupBy.join(","));
    if (opts?.from) params.set("from", opts.from);
    if (opts?.to) params.set("to", opts.to);
    if (opts?.tenant) params.set("tenant", opts.tenant);
    const q = params.toString() ? `?${params.toString()}` : "";
    return jsonFetch<UsageReportResponse>(this.ctx, `/v1/_usage${q}`, opts);
  }

  /** Instance configuration: build identity, the feature matrix, and the live
   *  provider/model/search cascade with coarse active/inactive status. The
   *  out-of-band twin of the in-band `Context op=capabilities`, sharing its
   *  probe server-side so the two cannot disagree. Mirrors `GET /v1/config`.
   *
   *  `view` names the disclosure level the response was rendered at, so you can
   *  tell "this deployment has none" from "you weren't shown them". A deployment
   *  running with `LOOMCYCLE_PUBLIC_CONFIG=1` also serves this **with no
   *  bearer**, returning the narrower `"public"` view — which is what a landing
   *  page fetches directly. `models` is one entry per (provider, model),
   *  deduplicated across plans, with `selected` marking what actually runs. */
  async getConfig(opts?: { signal?: AbortSignal }): Promise<ConfigResponse> {
    return jsonFetch<ConfigResponse>(this.ctx, "/v1/config", opts);
  }

  /** List the per-scope token budgets visible to the caller (RFC AW), each with
   *  its live month-to-date usage. Tenant-scoped server-side: a tenant operator
   *  sees only its own tenant's budgets; an admin sees all (or focuses one via
   *  `tenant`). Mirrors `GET /v1/_limits`. */
  async listLimits(opts?: {
    /** Admin-only tenant focus (?tenant=); ignored for a tenant principal. */
    tenant?: string;
    signal?: AbortSignal;
  }): Promise<TokenLimitsResponse> {
    const q = opts?.tenant ? `?tenant=${encodeURIComponent(opts.tenant)}` : "";
    return jsonFetch<TokenLimitsResponse>(this.ctx, `/v1/_limits${q}`, opts);
  }

  /** Upsert one token budget (RFC AW). A present `soft_limit`/`hard_limit` sets
   *  that tier; omitting it clears the tier (unlimited on that axis) — a
   *  full-row upsert. The operator-global scope + any cross-tenant `tenant_id`
   *  are admin-only (403 otherwise). Mirrors `PUT /v1/_limits`. */
  async setLimit(
    body: SetTokenLimitRequest,
    opts?: { signal?: AbortSignal },
  ): Promise<TokenLimit> {
    return putJSON<TokenLimit>(this.ctx, "/v1/_limits", body, opts);
  }

  /** Delete a token budget → the scope is unlimited again (RFC AW). Same
   *  tenant-confinement as `setLimit`. Mirrors `DELETE /v1/_limits`. */
  async deleteLimit(
    scope: string,
    opts?: {
      /** Required for scope=user (the subject); empty for scope=tenant. */
      scopeId?: string;
      /** Admin-only target tenant; ignored for a tenant principal. */
      tenant?: string;
      signal?: AbortSignal;
    },
  ): Promise<void> {
    const params = new URLSearchParams({ scope });
    if (opts?.scopeId) params.set("scope_id", opts.scopeId);
    if (opts?.tenant) params.set("tenant", opts.tenant);
    return deleteRequest(this.ctx, `/v1/_limits?${params.toString()}`, opts);
  }

  /** Read the full event log for a session. Each entry has seq,
   *  run_id, ts_ns, type, event (the providers.Event payload). */
  async getTranscript(
    sessionId: string,
    opts?: { signal?: AbortSignal },
  ): Promise<TranscriptResponse> {
    return jsonFetch<TranscriptResponse>(
      this.ctx,
      `/v1/sessions/${encodeURIComponent(sessionId)}/transcript`,
      opts,
    );
  }

  /** Liveness probe. Unauthenticated. Returns build info + uptime.
   *  Hits /healthz, not /v1/. */
  async health(opts?: { signal?: AbortSignal }): Promise<HealthResponse> {
    return jsonFetch<HealthResponse>(this.ctx, "/healthz", opts);
  }

  /** List known users with running-count summary. Drives the Web UI's
   *  user picker. Tenant-scoped server-side since v0.17.0 (RFC L): a
   *  tenant principal sees only its own tenant's users; an admin sees all,
   *  or focuses one tenant via `tenant` (?tenant=). */
  async listUsers(opts?: {
    tenant?: string;
    signal?: AbortSignal;
  }): Promise<ListUsersResponse> {
    const q = opts?.tenant ? `?tenant=${encodeURIComponent(opts.tenant)}` : "";
    return jsonFetch<ListUsersResponse>(this.ctx, `/v1/_users${q}`, opts);
  }

  /** Whoami — the authenticated principal (RFC L v0.17.0). Any
   *  authenticated bearer; returns its authoritative tenant / subject /
   *  scopes + `is_admin`. `open_mode: true` when the server runs without
   *  the token substrate (single shared LOOMCYCLE_AUTH_TOKEN). */
  async whoami(opts?: { signal?: AbortSignal }): Promise<WhoamiResponse> {
    return jsonFetch<WhoamiResponse>(this.ctx, "/v1/_me", opts);
  }

  // ---- v0.8.17/8.18 Pause / Resume / State ----

  /** Quiesce the runtime. Idempotent tools cancel immediately;
   *  non-idempotent + external tools get a grace window then
   *  force-cancel. Raises AlreadyPausingError on 409,
   *  PauseNotConfiguredError on 503. */
  async pauseRuntime(opts?: {
    timeoutMs?: number;
    signal?: AbortSignal;
  }): Promise<PauseResult> {
    const body =
      opts?.timeoutMs && opts.timeoutMs > 0
        ? { timeout_ms: opts.timeoutMs }
        : undefined;
    return postJSON<PauseResult>(this.ctx, "/v1/_pause", body, opts);
  }

  /** Release the runtime quiesce. Raises NotPausedError on 409. */
  async resumeRuntime(opts?: { signal?: AbortSignal }): Promise<ResumeResult> {
    return postJSON<ResumeResult>(this.ctx, "/v1/_resume", undefined, opts);
  }

  /** Current runtime state. Cheap query — atomic state + a
   *  bounded snapshots count. */
  async getRuntimeState(opts?: {
    signal?: AbortSignal;
  }): Promise<RuntimeStateResponse> {
    return jsonFetch<RuntimeStateResponse>(this.ctx, "/v1/_state", opts);
  }

  /** Trigger an immediate re-probe of every configured provider and
   *  return the refreshed availability matrix. Operator-only escape
   *  hatch when a transient outage stalls every provider and the
   *  runtime would otherwise 503 until the next periodic probe.
   *  Mirrors POST /v1/_resolve/probe. */
  async resolveProbe(opts?: {
    signal?: AbortSignal;
  }): Promise<ResolverMatrix> {
    return postJSON<ResolverMatrix>(
      this.ctx,
      "/v1/_resolve/probe",
      undefined,
      opts,
    );
  }

  // ---- Snapshot lifecycle ----

  /** Capture running-state into a per-section-semver JSON envelope.
   *  Raises SnapshotTooLargeError on 413 when the envelope exceeds
   *  LOOMCYCLE_SNAPSHOT_MAX_BYTES (default 512 MiB). */
  async createSnapshot(
    opts?: CreateSnapshotOptions & { signal?: AbortSignal },
  ): Promise<SnapshotCreateResponse> {
    const body: Record<string, unknown> = {};
    if (opts?.label) body.label = opts.label;
    if (opts?.includeHistory) body.include_history = true;
    if (opts?.includeHistorySince)
      body.include_history_since = opts.includeHistorySince;
    if (opts?.maxBytes && opts.maxBytes > 0) body.max_bytes = opts.maxBytes;
    return postJSON<SnapshotCreateResponse>(this.ctx, "/v1/_snapshots", body, opts);
  }

  /** List captured snapshots (most-recent first). Capped at 200
   *  server-side; the limit param defaults to 200 too. */
  async listSnapshots(opts?: {
    limit?: number;
    labelContains?: string;
    signal?: AbortSignal;
  }): Promise<SnapshotDescriptor[]> {
    const params = new URLSearchParams();
    if (opts?.limit && opts.limit > 0) params.set("limit", String(opts.limit));
    if (opts?.labelContains)
      params.set("label_contains", opts.labelContains);
    const qs = params.toString();
    const path = qs ? `/v1/_snapshots?${qs}` : "/v1/_snapshots";
    const resp = await jsonFetch<SnapshotListResponse>(this.ctx, path, opts);
    return resp.entries ?? [];
  }

  /** Fetch the full snapshot envelope including JSON content.
   *  Distinct from exportSnapshot (which is operator-facing
   *  "where did this land on the host" semantics with a download
   *  URL). Raises SnapshotNotFoundError on 404. */
  async getSnapshot(
    snapshotId: string,
    opts?: { signal?: AbortSignal },
  ): Promise<SnapshotEnvelope> {
    return jsonFetch<SnapshotEnvelope>(
      this.ctx,
      `/v1/_snapshots/${encodeURIComponent(snapshotId)}`,
      opts,
    );
  }

  /** Returns the URL of the snapshot's canonical envelope —
   *  synchronous and side-effect-free; does NOT issue an HTTP
   *  request. The endpoint is bearer-authed like every other
   *  `/v1/_snapshots/*` route, so callers must attach the same
   *  `Authorization: Bearer <token>` header when fetching this
   *  URL (e.g. `curl -H "Authorization: Bearer $TOKEN" ...`).
   *  There is no token query-param fallback. */
  exportSnapshotURL(snapshotId: string): string {
    return `${this.ctx.baseUrl}/v1/_snapshots/${encodeURIComponent(snapshotId)}/export`;
  }

  /** Open a client-tool host (RFC BC): a persistent WebSocket over which this
   *  client registers tools it runs on the user's machine + answers the
   *  agent's tool calls. Returns a started {@link ClientToolHost}; call
   *  `.close()` to stop. Uses the global WebSocket (browsers, Node 22+); pass
   *  `WebSocketImpl` (e.g. the `ws` package) on older Node. */
  connectClientTools(opts: ConnectClientToolsOptions): ClientToolHost {
    const host = new ClientToolHost(clientToolsURL(this.ctx.baseUrl), this.ctx.authToken, opts);
    host.start();
    return host;
  }

  /** Restore from a same-instance snapshot id OR an inline
   *  envelope JSON. Idempotent: ON CONFLICT DO NOTHING per row;
   *  the returned counters reflect rows actually written.
   *  Raises SnapshotVersionError on 422 when a section's
   *  declared version is newer than the reader supports. */
  async restoreSnapshot(opts: {
    snapshotId?: string;
    /** Pass a parsed JSON object (e.g. `getSnapshot.json_content` or
     *  the result of JSON.parse on an exported envelope). */
    json?: unknown;
    includeHistory?: boolean;
    signal?: AbortSignal;
  }): Promise<SnapshotRestoreResponse> {
    if (!opts.snapshotId && opts.json === undefined) {
      // Client-side validation — match Python adapter's
      // InvalidArgumentError pattern but the typed-error layer
      // lives in errors.ts; for a thrown plain error here the
      // method's catchers just see the message.
      throw new Error(
        "restoreSnapshot: pass snapshotId or json (one is required)",
      );
    }
    if (opts.snapshotId && opts.json !== undefined) {
      throw new Error(
        "restoreSnapshot: pass only one of snapshotId or json",
      );
    }
    // When json is supplied the id path-segment is ignored
    // server-side; we use a placeholder "inline" segment to keep
    // the URL well-formed.
    const id = opts.snapshotId ?? "inline";
    const body: Record<string, unknown> = {};
    if (opts.includeHistory) body.include_history = true;
    if (opts.json !== undefined) body.json = opts.json;
    return postJSON<SnapshotRestoreResponse>(
      this.ctx,
      `/v1/_snapshots/${encodeURIComponent(id)}/restore`,
      body,
      opts,
    );
  }

  /** Delete a snapshot. Idempotent — succeeds whether or not the
   *  row existed (server returns 204 in both cases). */
  async deleteSnapshot(
    snapshotId: string,
    opts?: { signal?: AbortSignal },
  ): Promise<void> {
    await deleteRequest(
      this.ctx,
      `/v1/_snapshots/${encodeURIComponent(snapshotId)}`,
      opts,
    );
  }

  // ---- Memory admin ----

  /** List the kinds of memory scopes the server knows about
   *  (agent, user — or whatever the operator yaml declares). */
  async listMemoryScopes(opts?: {
    signal?: AbortSignal;
  }): Promise<MemoryScopesResponse> {
    return jsonFetch<MemoryScopesResponse>(this.ctx, "/v1/_memory/scopes", opts);
  }

  /** List the scope_ids that have at least one memory row under
   *  a given scope. */
  async listMemoryScopeIDs(
    scope: string,
    opts?: { signal?: AbortSignal },
  ): Promise<MemoryScopeIDsResponse> {
    return jsonFetch<MemoryScopeIDsResponse>(
      this.ctx,
      `/v1/_memory/scopes/${encodeURIComponent(scope)}`,
      opts,
    );
  }

  /** List memory entries under a (scope, scope_id) tuple.
   *  Optional prefix narrows by key prefix. */
  async listMemoryEntries(
    scope: string,
    scopeID: string,
    opts?: { prefix?: string; limit?: number; signal?: AbortSignal },
  ): Promise<MemoryEntriesResponse> {
    const params = new URLSearchParams();
    if (opts?.prefix) params.set("prefix", opts.prefix);
    // Guard against `limit: 0` (falsy but valid-looking) and negatives —
    // both would either send `limit=0` (server treats as default but the
    // semantic is unclear) or `limit=-N` (server rejects). Only send the
    // param when the caller passed a meaningful positive number.
    if (opts?.limit && opts.limit > 0) params.set("limit", String(opts.limit));
    const qs = params.toString();
    const path = `/v1/_memory/scopes/${encodeURIComponent(scope)}/${encodeURIComponent(scopeID)}/keys${qs ? "?" + qs : ""}`;
    return jsonFetch<MemoryEntriesResponse>(this.ctx, path, opts);
  }

  /** Read a single memory entry by (scope, scope_id, key). */
  async getMemoryEntry(
    scope: string,
    scopeID: string,
    key: string,
    opts?: { signal?: AbortSignal },
  ): Promise<MemoryEntryResponse> {
    return jsonFetch<MemoryEntryResponse>(
      this.ctx,
      `/v1/_memory/scopes/${encodeURIComponent(scope)}/${encodeURIComponent(scopeID)}/keys/${encodeURIComponent(key)}`,
      opts,
    );
  }

  // ---- RFC BV memory-view: off-run search + embed-admin reads ----

  /** Off-run unified semantic search over one scope's memory
   *  (POST /v1/_memory/search). Spans BOTH plain k/v entries AND
   *  document-chunk bodies in one ranked list — each hit is tagged
   *  `kind: "memory" | "document"`, and a document hit carries
   *  `chunk_id` so the caller can fetch its entity block via
   *  document({ op: "get_chunk" }). The in-band Memory tool's `search`
   *  op is prefix-scoped + run-bound; this is its off-run twin for the
   *  admin/operator surface. HTTP-only (no gRPC RPC). `rank` / `dedup`
   *  are opaque ranking-config objects passed through as-is. */
  async memorySearch(
    input: MemorySearchInput,
    opts?: { signal?: AbortSignal },
  ): Promise<MemorySearchResponse> {
    const body: Record<string, unknown> = {
      query: input.query,
      scope: input.scope,
      scope_id: input.scopeId,
    };
    // Only send top_k when set — the server defaults + caps it (1..50).
    if (input.topK !== undefined) body.top_k = input.topK;
    if (input.rank !== undefined) body.rank = input.rank;
    if (input.dedup !== undefined) body.dedup = input.dedup;
    // Sent only when set, so an omitted selector keeps the endpoint's
    // span-everything default rather than encoding one here (RFC BW).
    if (input.sources !== undefined) body.sources = input.sources;
    return postJSON<MemorySearchResponse>(this.ctx, "/v1/_memory/search", body, {
      signal: opts?.signal,
    });
  }

  /** Per-(provider, model, dimension) row counts + total embedding
   *  bytes for one scope (GET /v1/_memory/embed_stats). The memory-view
   *  console shows this to spot a multi-embedder scope BEFORE a reembed
   *  migration. */
  async memoryEmbedStats(
    scope: string,
    opts?: { signal?: AbortSignal },
  ): Promise<MemoryEmbedStatsResponse> {
    return jsonFetch<MemoryEmbedStatsResponse>(
      this.ctx,
      `/v1/_memory/embed_stats?scope=${encodeURIComponent(scope)}`,
      opts,
    );
  }

  /** Re-embed a scope's rows whose stored embedder differs from the
   *  configured one (POST /v1/_memory/reembed). `dry_run` defaults TRUE
   *  server-side, so an omitted `dryRun` returns the planned migration
   *  WITHOUT writing; pass `dryRun: false` to commit. `limit` caps the
   *  rows processed per call (server default 1000). The response is a
   *  discriminated union on `dry_run` — narrow before reading its
   *  arm-specific fields. */
  async reembedMemory(
    scope: string,
    scopeId: string,
    opts?: { dryRun?: boolean; limit?: number; signal?: AbortSignal },
  ): Promise<MemoryReembedResponse> {
    const query: Record<string, string> = { scope, scope_id: scopeId };
    // Only send dry_run=false to commit; an omitted flag leaves the
    // server-side default (true) so a caller can't accidentally reembed.
    if (opts?.dryRun === false) query.dry_run = "false";
    if (opts?.limit !== undefined && opts.limit > 0) {
      query.limit = String(opts.limit);
    }
    return postJSON<MemoryReembedResponse>(
      this.ctx,
      "/v1/_memory/reembed",
      undefined,
      { query, signal: opts?.signal },
    );
  }

  // ---- Interruption ----

  /** List interrupts addressable to a user_id. Default filter is
   *  status=pending. */
  async listUserInterrupts(
    userId: string,
    opts?: { status?: InterruptStatus; signal?: AbortSignal },
  ): Promise<InterruptListResponse> {
    const status = opts?.status ?? "pending";
    return jsonFetch<InterruptListResponse>(
      this.ctx,
      `/v1/users/${encodeURIComponent(userId)}/interrupts?status=${encodeURIComponent(status)}`,
      opts,
    );
  }

  /** List interrupts emitted by a specific run. */
  async listRunInterrupts(
    runId: string,
    opts?: { status?: InterruptStatus; signal?: AbortSignal },
  ): Promise<InterruptListResponse> {
    const status = opts?.status ?? "pending";
    return jsonFetch<InterruptListResponse>(
      this.ctx,
      `/v1/runs/${encodeURIComponent(runId)}/interrupts?status=${encodeURIComponent(status)}`,
      opts,
    );
  }

  /** Resolve a pending Interruption.ask from outside the agent
   *  loop. Lets a TS-side dashboard or service act as the human
   *  answerer when operator yaml configures the consumer-MCP
   *  backend. Pass `disposition: "declined"` (or use
   *  {@link LoomcycleClient.cancelInterrupt}) to decline WITHOUT an
   *  answer so the waiting Question tool proceeds (RFC BH). */
  async resolveInterrupt(
    runId: string,
    interruptId: string,
    opts: ResolveInterruptOptions & { signal?: AbortSignal },
  ): Promise<unknown> {
    const body: Record<string, unknown> = {
      kind: opts.kind ?? "question",
      resolved_by: opts.resolvedBy ?? "client",
    };
    // A decline carries no answer; omit it rather than sending "" so the
    // server's "declined must not carry an answer" gate isn't tripped.
    if (opts.answer !== undefined) body.answer = opts.answer;
    if (opts.disposition !== undefined) body.disposition = opts.disposition;
    return postJSON<unknown>(
      this.ctx,
      `/v1/runs/${encodeURIComponent(runId)}/interrupts/${encodeURIComponent(interruptId)}/resolve`,
      body,
      opts,
    );
  }

  /** Decline a pending Interruption.ask WITHOUT answering it (RFC BH) — the
   *  waiting Question tool returns a non-error "declined" result so the agent
   *  proceeds, and the run keeps going. A thin wrapper over
   *  {@link LoomcycleClient.resolveInterrupt} with `disposition: "declined"`
   *  (no answer). Mirrors POST .../resolve `{ disposition: "declined" }`. */
  async cancelInterrupt(
    runId: string,
    interruptId: string,
    opts?: { resolvedBy?: string; signal?: AbortSignal },
  ): Promise<unknown> {
    return this.resolveInterrupt(runId, interruptId, {
      disposition: "declined",
      resolvedBy: opts?.resolvedBy,
      signal: opts?.signal,
    });
  }

  // ---- Hook management (hooks-connector series, PR C) ----

  /** Register a pre- or post-tool webhook. The callback_url must be
   *  an http:// or https:// endpoint the CONSUMER runs — loomcycle
   *  POSTs PreHookCall / PostHookCall payloads to it. This method
   *  manages registration only; the receiver is the consumer's own
   *  HTTP framework (Express, Next.js, etc.).
   *
   *  Re-registering the same (owner, name) replaces the prior entry
   *  with a fresh id (idempotent app-restart contract).
   *
   *  Raises InvalidArgumentError on 400 (bad URL / phase / missing
   *  required fields). */
  async registerHook(
    opts: RegisterHookOptions & { signal?: AbortSignal },
  ): Promise<RegisterHookResponse> {
    const body: Record<string, unknown> = {
      owner: opts.owner,
      name: opts.name,
      phase: opts.phase,
      callback_url: opts.callbackUrl,
    };
    if (opts.agents !== undefined) body.agents = opts.agents;
    if (opts.tools !== undefined) body.tools = opts.tools;
    if (opts.failMode !== undefined) body.fail_mode = opts.failMode;
    if (opts.timeoutMs !== undefined && opts.timeoutMs > 0) {
      body.timeout_ms = opts.timeoutMs;
    }
    return postJSON<RegisterHookResponse>(this.ctx, "/v1/hooks", body, opts);
  }

  /** List every currently-registered hook. Returns the array
   *  unwrapped (the wire envelope is `{hooks: [...]}` — we strip
   *  the envelope to match listUserAgents). In-memory only — empty
   *  after a loomcycle restart. */
  async listHooks(opts?: { signal?: AbortSignal }): Promise<Hook[]> {
    const resp = await jsonFetch<ListHooksResponse>(this.ctx, "/v1/hooks", opts);
    return resp.hooks ?? [];
  }

  /** Delete a registered hook by id. Raises HookNotFoundError on
   *  404. Returns void on success (the HTTP 200 body `{deleted: id}`
   *  is dropped — callers already know the id they passed). */
  async deleteHook(id: string, opts?: { signal?: AbortSignal }): Promise<void> {
    await deleteRequest(
      this.ctx,
      `/v1/hooks/${encodeURIComponent(id)}`,
      opts,
    );
  }

  // ---- v0.8.22 substrate admin (AgentDef + SkillDef) ----

  /** Invoke the AgentDef substrate tool over HTTP. Mirrors the
   *  MCP `agentdef` meta-tool and the in-band agent tool_use of
   *  the same name — different transport, identical semantics.
   *
   *  The `input.op` field discriminates create / fork / get /
   *  list / promote / retire. The remaining fields are op-specific;
   *  see the in-process tool's documentation.
   *
   *  Raises {@link SubstrateToolRefusedError} when the tool itself
   *  refuses the call (scope deny, empty body, allowed-tools
   *  widening, etc.) — distinct from transport failures so callers
   *  can branch on the typed error class.
   *
   *  Raises {@link InvalidArgumentError} on 400 (malformed JSON
   *  body); {@link AuthError} on 401; {@link UnavailableError} on
   *  503 (store / connector unwired). */
  async agentDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_agentdef", input, opts);
  }

  /** Register (or refresh) a code-js agent idempotently from an INLINE JS body
   *  — the typed "ship a code agent with no host filesystem" convenience
   *  (RFC J), mirror of {@link LoomcycleClient.ensureMcpServer}.
   *
   *  Runs `agentDef({op:"create", overlay:{provider:"code-js", code_body, …}})`,
   *  which is content-addressed-idempotent in loomcycle: a byte-identical
   *  re-registration is a no-op (`changed:false`), not a new version. This is
   *  what makes code agents work across container boundaries / pure-cloud —
   *  the JS no longer has to exist at `agent_code/<name>/index.js` on the
   *  sidecar's disk.
   *
   *  Requires `LOOMCYCLE_CODE_AGENTS_ENABLED=1` on the sidecar — the same
   *  switch that registers the code-js provider. Without it the create is
   *  refused (raises {@link SubstrateToolRefusedError}). Keep `${run.*}` /
   *  `${LOOMCYCLE_*}` placeholders in the body LITERAL so the content stays
   *  stable and dedup engages on re-register. */
  async ensureCodeAgent(
    opts: EnsureCodeAgentOptions,
    callOpts?: { signal?: AbortSignal },
  ): Promise<EnsureCodeAgentResult> {
    const overlay: AgentDefOverlay = { provider: "code-js", code_body: opts.code };
    if (opts.tools) overlay.tools = opts.tools;
    if (opts.tier) overlay.tier = opts.tier;
    if (opts.model) overlay.model = opts.model;
    const input: SubstrateToolInput = { op: "create", name: opts.name, overlay };
    if (opts.description) input.description = opts.description;

    const row = (await this.agentDef(input, callOpts)) as AgentDefRowResponse;
    return {
      name: opts.name,
      defId: row.def_id,
      version: row.version,
      changed: row.deduplicated !== true,
    };
  }

  /** Invoke the SkillDef substrate tool over HTTP. Mirror of
   *  {@link LoomcycleClient.agentDef} for skills (v0.8.22+). Same
   *  input grammar, same error class on refusal. See the
   *  agentDef() doc for the full shape and error contract. */
  async skillDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_skilldef", input, opts);
  }

  /** Invoke the v0.9.x MCPServerDef substrate tool over HTTP.
   *  Dynamic MCP server registration — register an HTTP /
   *  Streamable-HTTP MCP server at runtime so its tools become
   *  callable from any agent's `tools` list without a yaml
   *  edit + restart.
   *
   *  Operator-admin-only: this endpoint requires the bearer token.
   *
   *  Op-discriminated input: `{op: "create" | "fork" | "get" | "list"
   *  | "promote" | "retire" | "rediscover" | "verify", ...}`. Returns
   *  shape varies — narrow with {@link MCPServerDefRowResponse} for
   *  create/fork/get/list rows, {@link MCPServerDefVerifyResult} for
   *  verify responses.
   *
   *  Hard constraints (substrate refuses these):
   *  - Transport must be `http` or `streamable-http` (stdio stays
   *    yaml-only — dynamic registration doesn't allow process spawn).
   *  - URL hostname must be in LOOMCYCLE_HTTP_HOST_ALLOWLIST or (since
   *    v0.17.x) LOOMCYCLE_HTTP_PRIVATE_HOST_ALLOWLIST — the latter is where
   *    a self-hosted loopback callback like `http://localhost:3000/api/mcp`
   *    belongs (SSRF defence at the registration boundary).
   *  - Name colliding with a static cfg.MCPServers entry is refused
   *    (yaml is ground truth; use a different name).
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals
   *  (transport/host/yaml-name); {@link InvalidArgumentError} on 400
   *  (malformed JSON); {@link AuthError} on 401. */
  async mcpServerDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_mcpserverdef", input, opts);
  }

  /** Typed `MCPServerDef verify` — "is `contentSha256` the active version's
   *  hash for `name`?" `matches: true` is the no-op signal (the analog of
   *  the agent/skill verify-before-create dedup). Thin typed wrapper over
   *  {@link mcpServerDef}. */
  async mcpServerDefVerify(
    name: string,
    contentSha256: string,
    opts?: { signal?: AbortSignal },
  ): Promise<MCPServerDefVerifyResult> {
    return (await this.mcpServerDef(
      { op: "verify", name, content_sha256: contentSha256 },
      opts,
    )) as MCPServerDefVerifyResult;
  }

  /** Register (or refresh) a dynamic MCP server idempotently — the typed
   *  "register-if-changed" convenience for a consumer that re-registers its
   *  own callback server on every startup.
   *
   *  Runs `create`, which is content-addressed-idempotent (a byte-identical
   *  re-registration is a no-op, not a new version) AND auto-discovers the
   *  upstream's tools at ingestion — so {@link EnsureMcpServerResult.changed}
   *  is false on a stable-content re-register, and
   *  {@link EnsureMcpServerResult.discoveredToolCount} is populated straight
   *  from the `create` (no separate rediscover needed for the count). On a
   *  deduplicated re-register the count is absent (the tool surface was
   *  unchanged), so it reflects only the freshly-discovered case.
   *
   *  Pass {@link EnsureMcpServerOptions.rediscover} only to FORCE a refresh
   *  when the upstream's tools changed but the registration content didn't.
   *
   *  Keep `${run.*}` / `${LOOMCYCLE_*}` header placeholders LITERAL (don't
   *  resolve a per-restart token) or the content varies each boot and dedup
   *  can't engage. */
  async ensureMcpServer(
    opts: EnsureMcpServerOptions,
    callOpts?: { signal?: AbortSignal },
  ): Promise<EnsureMcpServerResult> {
    const overlay: Record<string, unknown> = {
      transport: opts.transport ?? "http",
      url: opts.url,
    };
    if (opts.headers) overlay.headers = opts.headers;
    const createInput: SubstrateToolInput = { op: "create", name: opts.name, overlay };
    if (opts.description) createInput.description = opts.description;

    const created = (await this.mcpServerDef(createInput, callOpts)) as MCPServerDefRowResponse;
    let row = created;
    let changed = created.deduplicated !== true;
    // create auto-discovers at ingestion → the count is on the create response.
    let discoveredToolCount = created.discovered;

    if (opts.rediscover) {
      const red = (await this.mcpServerDef(
        { op: "rediscover", name: opts.name },
        callOpts,
      )) as MCPServerDefRowResponse;
      row = red;
      if (red.deduplicated !== true) changed = true;
      if (red.discovered !== undefined) discoveredToolCount = red.discovered;
    }

    const result: EnsureMcpServerResult = {
      name: opts.name,
      defId: row.def_id,
      version: row.version,
      changed,
    };
    if (discoveredToolCount !== undefined) result.discoveredToolCount = discoveredToolCount;
    return result;
  }

  /** Invoke the v1.x RFC E ScheduleDef substrate tool over HTTP.
   *  Author + fork + retire scheduled-run definitions at runtime.
   *  Mirror of {@link LoomcycleClient.agentDef} for schedules — the
   *  primary use case is JobEmber-style "fork a yaml template
   *  per-user with their bearer + tier cron" workflows.
   *
   *  Operator-admin-only: this endpoint requires the bearer token.
   *
   *  Op-discriminated input: `{op: "create" | "fork" | "get" |
   *  "list" | "retire", ...}`. Note that ScheduleDef has 5 ops
   *  (no separate `promote` — forks auto-promote by default per
   *  RFC E's worked example; no `verify` — no content_sha256 in
   *  v1.x).
   *
   *  Sharp edges (substrate refuses these):
   *  - Name colliding with a static cfg.ScheduledRuns entry is
   *    refused on `create` (yaml is ground truth; use `fork` to
   *    derive a new version).
   *  - Fork against a template with `required_credentials` must
   *    supply all keys in `user_credentials` — loud-fail at fork
   *    time rather than silent ingestion-time failure when the
   *    sweeper fires.
   *  - Cron syntax is validated server-side; invalid expressions
   *    refuse with a parse error.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals
   *  (static-name collision, missing required credential, invalid
   *  cron, etc.); {@link InvalidArgumentError} on 400 (malformed
   *  JSON); {@link AuthError} on 401. */
  async scheduleDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_scheduledef", input, opts);
  }

  /** Invoke the v1.x RFC G A2AServerCardDef substrate tool over HTTP.
   *  Author + fork + retire A2A server-card definitions at runtime.
   *  Mirror of {@link LoomcycleClient.scheduleDef} for A2A server cards.
   *
   *  Operator-admin-only: this endpoint requires the bearer token.
   *
   *  Op-discriminated input: `{op: "create" | "fork" | "get" |
   *  "list" | "retire", ...}`.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals;
   *  {@link InvalidArgumentError} on 400 (malformed JSON);
   *  {@link AuthError} on 401. */
  async a2aServerCardDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_a2aservercarddef", input, opts);
  }

  /** Invoke the v1.x RFC G A2AAgentDef substrate tool over HTTP.
   *  Author + fork + retire A2A agent definitions at runtime.
   *  Mirror of {@link LoomcycleClient.scheduleDef} for A2A agents.
   *
   *  Operator-admin-only: this endpoint requires the bearer token.
   *
   *  Op-discriminated input: `{op: "create" | "fork" | "get" |
   *  "list" | "retire", ...}`.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals;
   *  {@link InvalidArgumentError} on 400 (malformed JSON);
   *  {@link AuthError} on 401. */
  async a2aAgentDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_a2aagentdef", input, opts);
  }

  /** Invoke the v1.x RFC H WebhookDef substrate tool over HTTP.
   *  Author + fork + retire inbound webhook definitions at runtime.
   *  Mirror of {@link LoomcycleClient.a2aAgentDef} for inbound webhooks.
   *
   *  Operator-admin-only: this endpoint requires the bearer token.
   *
   *  Op-discriminated input: `{op: "create" | "fork" | "get" |
   *  "list" | "retire", ...}`.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals;
   *  {@link InvalidArgumentError} on 400 (malformed JSON);
   *  {@link AuthError} on 401. */
  async webhookDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_webhookdef", input, opts);
  }

  /** Invoke the RFC I MR-3a MemoryBackendDef substrate tool over HTTP.
   *  Author + fork + retire named memory backend definitions at runtime.
   *  Mirror of {@link LoomcycleClient.webhookDef} for memory backends.
   *
   *  Operator-admin-only: this endpoint requires the bearer token.
   *
   *  Op-discriminated input: `{op: "create" | "fork" | "get" |
   *  "list" | "retire", ...}`.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals;
   *  {@link InvalidArgumentError} on 400 (malformed JSON);
   *  {@link AuthError} on 401. */
  async memoryBackendDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_memorybackenddef", input, opts);
  }

  /** OperatorTokenDef substrate ops (create/rotate/retire/get/list) —
   *  RFC L OSS multi-tenant authorization. Operator-admin only. The
   *  token plaintext is returned ONCE in the create/rotate response.
   *  Mirrors POST /v1/_operatortokendef. */
  async operatorTokenDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_operatortokendef", input, opts);
  }

  // ---- RFC AH (v0.35.0) dynamic filesystem volumes ----

  /** Invoke the RFC AH VolumeDef substrate tool over HTTP. Provision +
   *  manage CONFINED, per-tenant filesystem volumes at runtime.
   *  Tenant-confined (ScopeTenant, not operator-admin-only): the runtime
   *  stamps the caller's authoritative tenant and opaque-404s cross-tenant.
   *
   *  Op-discriminated input. A VolumeDef is FLAT (a pointer to mutable
   *  on-disk state, not a versioned definition), so the op set is
   *  `{op: "create" | "get" | "list" | "delete" | "purge", ...}` — NOT the
   *  content-addressed retire/promote/fork of the other Def families:
   *  - `create` — `{op, name, mode}`. The runtime DERIVES the path inside the
   *    operator-blessed `dynamic_root` (`<root>/<tenant>/<name>`) and
   *    `MkdirAll`s it; you never supply a host path.
   *  - `delete` — non-destructive: drops the mapping, leaves files on disk.
   *  - `purge`  — destructive: drops the mapping AND `RemoveAll`s the tree.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals
   *  (collision with a static volume name, no `dynamic_root` configured,
   *  cross-tenant); {@link InvalidArgumentError} on 400; {@link AuthError}
   *  on 401. */
  async volumeDef(
    input: SubstrateToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<SubstrateToolResponse> {
    return postJSON<SubstrateToolResponse>(this.ctx, "/v1/_volumedef", input, opts);
  }

  // ---- RFC AP Agent Teams (TeamDef substrate) ----
  //
  // TeamDefs are agent-team workflow graphs (states + transitions + a per-state
  // handler agent). These mirror the HTTP shapes the Web UI drives; the
  // substrate is tenant-confined (the runtime stamps the caller's authoritative
  // tenant, opaque-404s cross-tenant). All op-dispatch through POST /v1/_teamdef
  // except listTeams, which reads GET /v1/_teamdef/names.

  /** List every team the caller may see (GET /v1/_teamdef/names), each with its
   *  version roll-up. Tenant-scoped server-side. `names` is `null` when empty. */
  async listTeams(opts?: { signal?: AbortSignal }): Promise<ListTeamsResponse> {
    return jsonFetch<ListTeamsResponse>(this.ctx, "/v1/_teamdef/names", opts);
  }

  /** Render a team's active version as a Mermaid `stateDiagram-v2` (op=
   *  render_diagram). `highlightState` optionally marks one state with a bold
   *  outline (e.g. a chunk's current state). */
  async renderTeamDiagram(
    name: string,
    opts?: { signal?: AbortSignal; highlightState?: string },
  ): Promise<TeamDiagram> {
    const body: Record<string, unknown> = { op: "render_diagram", name };
    if (opts?.highlightState) body.highlight_state = opts.highlightState;
    return postJSON<TeamDiagram>(this.ctx, "/v1/_teamdef", body, opts);
  }

  /** Fetch one version's full record incl. the editable `definition` graph
   *  (op=get by def_id). */
  async getTeamDef(defId: string, opts?: { signal?: AbortSignal }): Promise<TeamDefDetail> {
    return postJSON<TeamDefDetail>(this.ctx, "/v1/_teamdef", { op: "get", def_id: defId }, opts);
  }

  /** Author a new TeamDef (op=create). `overlay` is the workflow graph
   *  (`{entry, states, transitions, max_iterations?, colors?}`), validated
   *  server-side — an invalid graph raises {@link SubstrateToolRefusedError}. */
  async createTeam(
    name: string,
    overlay: Record<string, unknown>,
    opts?: { signal?: AbortSignal },
  ): Promise<CreatedTeam> {
    return postJSON<CreatedTeam>(this.ctx, "/v1/_teamdef", { op: "create", name, overlay }, opts);
  }

  /** Save an edited graph as a NEW version of an existing team (op=fork). With
   *  no parent the server forks the name's active def; `overlay` (a full graph)
   *  replaces it. Validated server-side ({@link SubstrateToolRefusedError}). */
  async forkTeam(
    name: string,
    overlay: Record<string, unknown>,
    opts?: { signal?: AbortSignal },
  ): Promise<CreatedTeam> {
    return postJSON<CreatedTeam>(this.ctx, "/v1/_teamdef", { op: "fork", name, overlay }, opts);
  }

  /** Hard-remove a whole team by name — all versions + the active pointer
   *  (op=delete), scoped to the caller's tenant. Teams are runtime-only, so
   *  this is how an operator clears an obsolete/test team. */
  async deleteTeam(
    name: string,
    opts?: { signal?: AbortSignal },
  ): Promise<{ name: string; deleted: boolean }> {
    return postJSON<{ name: string; deleted: boolean }>(
      this.ctx,
      "/v1/_teamdef",
      { op: "delete", name },
      opts,
    );
  }

  /** Execute a team (op=run) — walk its state graph, spawning each state's agent
   *  until a terminal state, returning the per-state trace. Target the active
   *  version by `name` OR a specific version by `defId`. `input` is the initial
   *  task handed to the entry state's agent. The walk runs under the same
   *  admission a normal run gets (token budget / operator-key / depth). */
  async runTeam(
    target: { name?: string; defId?: string; input?: string },
    opts?: { signal?: AbortSignal },
  ): Promise<TeamRunResult> {
    const body: Record<string, unknown> = { op: "run" };
    if (target.name !== undefined) body.name = target.name;
    if (target.defId !== undefined) body.def_id = target.defId;
    if (target.input !== undefined) body.input = target.input;
    return postJSON<TeamRunResult>(this.ctx, "/v1/_teamdef", body, opts);
  }

  /** Invoke the RFC AL Path VFS tool over HTTP (`POST /v1/_path`). A
   *  Unix-like filesystem over your Memory entries, Volume mounts, and
   *  Documents — address them by human-readable paths (e.g. /docs/launch).
   *  Op-discriminated (`resolve`/`ls`/`stat`/`mkdir`/`mv`/`rm`). Scope
   *  (agent/user/tenant) + tenant are resolved server-side from the
   *  authenticated principal, never the wire.
   *
   *  Response varies per op — an `ls` returns `{path, entries}`, a `resolve`
   *  returns the dirent + resource_ref. Narrow as needed.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals (bad
   *  path, rm of a non-empty path without recursive, etc.);
   *  {@link InvalidArgumentError} on 400; {@link AuthError} on 401.
   *
   *  `opts.scopeId` / `opts.tenant` are the RFC AS browse-by-subject overrides
   *  (sent as `?scope_id=` / `?tenant=` query params — the server reads them
   *  from the URL, not the body, and re-checks authorization: a tenant
   *  principal's `tenant` is ignored, `scopeId` picks any subject it may see).
   *  Omit both to browse your own subject (byte-identical to the pre-RFC-AS
   *  request). */
  async path(
    input: PathToolInput,
    opts?: { signal?: AbortSignal; scopeId?: string; tenant?: string },
  ): Promise<PathToolResponse> {
    return postJSON<PathToolResponse>(this.ctx, "/v1/_path", input, {
      signal: opts?.signal,
      query: browseQuery(opts),
    });
  }

  /** Invoke the RFC AK Document tool over HTTP (`POST /v1/_document`). A
   *  chunked-graph document where each chunk is a first-class unit (UUID,
   *  hierarchy, type, fields, graph edges, Markdown body) that agents and
   *  humans co-author. Op-discriminated (16 ops: document/chunk lifecycle,
   *  set_path, edges, query_chunks, type defs, export_md/import_md). Scope
   *  agent/user (tenant deferred); resolved server-side from the principal.
   *
   *  Requires SQL Memory enabled on the sidecar (`LOOMCYCLE_SQLMEM_ENABLED=1`)
   *  — the chunk-structure tables live there. Without it the call is refused
   *  ({@link SubstrateToolRefusedError}).
   *
   *  Response varies per op — `create_document` returns
   *  `{document_id, root_chunk_id, ...}`, `query_chunks` returns rows.
   *  {@link InvalidArgumentError} on 400; {@link AuthError} on 401.
   *
   *  `opts.scopeId` / `opts.tenant` are the RFC AS browse-by-subject overrides —
   *  see {@link LoomcycleClient.path}. Omit both to browse your own subject. */
  async document(
    input: DocumentToolInput,
    opts?: { signal?: AbortSignal; scopeId?: string; tenant?: string },
  ): Promise<DocumentToolResponse> {
    return postJSON<DocumentToolResponse>(this.ctx, "/v1/_document", input, {
      signal: opts?.signal,
      query: browseQuery(opts),
    });
  }

  /** Build the URL of an image chunk's asset (RFC BO) — `GET
   *  /v1/_document/asset/{chunkId}`. Attach the bytes to a chunk with the
   *  `set_asset` op via {@link LoomcycleClient.document}, then read them here.
   *  `scope` is "agent" | "user" | "tenant" (default "user"); `scopeId`/`tenant` are the
   *  RFC AS browse overrides. NOTE: the URL needs the bearer to fetch — use
   *  {@link LoomcycleClient.fetchDocumentAsset} for an authenticated GET (a bare
   *  <img src> can't carry the Authorization header). */
  documentAssetUrl(
    chunkId: string,
    opts?: { scope?: string; scopeId?: string; tenant?: string },
  ): string {
    const q = new URLSearchParams({ scope: opts?.scope ?? "user" });
    if (opts?.scopeId) q.set("scope_id", opts.scopeId);
    if (opts?.tenant) q.set("tenant", opts.tenant);
    return `${this.ctx.baseUrl}/v1/_document/asset/${encodeURIComponent(chunkId)}?${q.toString()}`;
  }

  /** Fetch an image chunk's asset bytes (RFC BO) with authentication — returns
   *  the raw {@link Response} so a caller can `.blob()` / `.arrayBuffer()` it.
   *  Adds the bearer header (a bare <img src> can't). Throws on a non-2xx
   *  (a cross-scope / missing asset is an opaque 404). */
  async fetchDocumentAsset(
    chunkId: string,
    opts?: { scope?: string; scopeId?: string; tenant?: string; signal?: AbortSignal },
  ): Promise<Response> {
    const headers: Record<string, string> = {};
    if (this.ctx.authToken) headers.Authorization = `Bearer ${this.ctx.authToken}`;
    const resp = await this.ctx.fetchImpl(this.documentAssetUrl(chunkId, opts), {
      method: "GET",
      headers,
      signal: opts?.signal,
    });
    if (!resp.ok) {
      throw new Error(`fetchDocumentAsset: ${resp.status} ${resp.statusText}`);
    }
    return resp;
  }

  /** Invoke the RFC BE History tool over HTTP (`POST /v1/_history`). Browse,
   *  search, and annotate PAST CHATS — a chat is a session (it may span
   *  several runs). Op-discriminated (`list`/`get`/`search`/`rename`/
   *  `annotate`/`pin`/`archive`/`recap`/`resume`).
   *
   *  The owner is resolved server-side from the authenticated principal, NEVER
   *  the wire: `scope:"user"` keys on your own subject, `scope:"tenant"` on your
   *  tenant. The cross-tenant `scope:"global"` is admin-only — a tenant operator
   *  requesting it is refused ({@link SubstrateToolRefusedError}).
   *
   *  Response varies per op — `list`/`search` return `{scope, chats, total}`,
   *  `get` returns the chat metadata + transcript (or Markdown when
   *  `format:"markdown"`), `recap` returns a summary. Narrow as needed.
   *
   *  Raises {@link SubstrateToolRefusedError} on tool-level refusals (unknown
   *  session, disallowed scope); {@link InvalidArgumentError} on 400;
   *  {@link AuthError} on 401. */
  async history(
    input: HistoryToolInput,
    opts?: { signal?: AbortSignal },
  ): Promise<HistoryToolResponse> {
    return postJSON<HistoryToolResponse>(this.ctx, "/v1/_history", input, opts);
  }

  /** List the PERSISTENT volume universe for the caller's tenant — every
   *  static volume (the shared bind floor, `source: "static"`, read-only)
   *  plus the tenant's own dynamic VolumeDefs (`source: "dynamic"`). Host
   *  paths are redacted (`""`) for a non-operator (tenant) caller.
   *
   *  Wraps `GET /v1/_volumes` (RFC AH Phase 4). Bearer-authed; tenant-scoped
   *  from the authoritative principal. */
  async listVolumes(opts?: {
    signal?: AbortSignal;
  }): Promise<PersistentVolumesResponse> {
    return jsonFetch<PersistentVolumesResponse>(this.ctx, "/v1/_volumes", opts);
  }

  /** List the LIVE ephemeral (run-scoped) volumes for the caller's tenant —
   *  created mid-run, inherited by sub-agents, auto-purged when the run
   *  completes. Host paths are redacted (`""`) for a non-operator caller.
   *
   *  Wraps `GET /v1/_volumes/ephemeral` (RFC AH Phase 4). Bearer-authed;
   *  tenant-scoped from the authoritative principal. */
  async listEphemeralVolumes(opts?: {
    signal?: AbortSignal;
  }): Promise<EphemeralVolumesResponse> {
    return jsonFetch<EphemeralVolumesResponse>(this.ctx, "/v1/_volumes/ephemeral", opts);
  }

  // ---- v0.10.3 Library v2 enumeration (read-only, merged yaml+substrate) ----

  /** List every agent the runtime knows about — yaml-static + dynamic
   *  AgentDefs merged into one envelope per name. Each entry carries
   *  `source: "static-only" | "dynamic-only" | "both"` so callers can
   *  badge / filter by origin. Static-source entries inline the
   *  yaml-side definition body in `static_definition`; dynamic-only
   *  entries omit it (fetch the active version's body via
   *  `agentDef({op: "get", name})` when needed).
   *
   *  Wraps the v0.9.3 endpoint `GET /v1/_library/agents`. Bearer-authed.
   *  Deterministic alphabetical sort by name.
   *
   *  Typical use: populate an integration's "which agent to dispatch?"
   *  dropdown (n8n, Slack slash commands, custom workflow editors).
   *
   *  Raises {@link AuthError} on 401; {@link UnavailableError} on 503
   *  (store unwired in test configurations). */
  async listLibraryAgents(opts?: {
    signal?: AbortSignal;
  }): Promise<LibraryListResponse<LibraryAgentDefinition>> {
    return jsonFetch<LibraryListResponse<LibraryAgentDefinition>>(
      this.ctx,
      "/v1/_library/agents",
      opts,
    );
  }

  /** List every skill the runtime knows about — bundled skills (Approach
   *  A) + filesystem overlay + dynamic SkillDefs merged into one
   *  envelope per name. See {@link LoomcycleClient.listLibraryAgents}
   *  for the source-tagging contract; the only difference is the
   *  `static_definition` shape ({@link LibrarySkillDefinition}).
   *
   *  Wraps `GET /v1/_library/skills` (v0.9.3). */
  async listLibrarySkills(opts?: {
    signal?: AbortSignal;
  }): Promise<LibraryListResponse<LibrarySkillDefinition>> {
    return jsonFetch<LibraryListResponse<LibrarySkillDefinition>>(
      this.ctx,
      "/v1/_library/skills",
      opts,
    );
  }

  /** List every MCP server the runtime knows about — yaml-declared
   *  servers + dynamic MCPServerDef registrations merged into one
   *  envelope per name. Static entries additionally inline the cached
   *  `discovered_tools` snapshot when the pool inspector is wired
   *  AND the pool has finished init for that server (omitted otherwise
   *  — re-check after the pool finishes).
   *
   *  Wraps `GET /v1/_library/mcp-servers` (v0.9.3). */
  async listLibraryMcpServers(opts?: {
    signal?: AbortSignal;
  }): Promise<LibraryListResponse<LibraryMcpServerDefinition>> {
    return jsonFetch<LibraryListResponse<LibraryMcpServerDefinition>>(
      this.ctx,
      "/v1/_library/mcp-servers",
      opts,
    );
  }

  // ---- v0.11.0 LLM Gateway ----

  /** Non-streaming LLM chat completion via the gateway endpoint.
   *  Wraps `POST /v1/_llm/chat` with `stream: false`. The gateway
   *  resolves a provider per the routing precedence
   *  (explicit-pin > explicit-provider > explicit-model > resolver
   *  default), invokes it directly (no agent loop), and returns the
   *  aggregated response with usage counters + the chosen
   *  provider/model echoed back.
   *
   *  Use this when you want loomcycle's routing benefits (one
   *  credential, one quota, one observability surface across
   *  providers) without paying for the agent runtime overhead.
   *  Tool-calling works: pass `tools[]` and read `tool_use` content
   *  blocks back; the per-provider schema translation is handled by
   *  the substrate's existing driver layer.
   *
   *  Raises {@link AuthError} on 401; {@link UnavailableError} on 503
   *  (resolver not configured, store unwired);
   *  {@link InvalidArgumentError} on 400 (bad request shape). */
  async llmChat(opts: LLMChatOptions): Promise<LLMChatResponse> {
    const body = serializeLLMOptions(opts, false);
    return postJSON<LLMChatResponse>(this.ctx, "/v1/_llm/chat", body, {
      signal: opts.signal,
    });
  }

  /** Streaming LLM chat completion. Wraps `POST /v1/_llm/chat` with
   *  `stream: true`. Yields one {@link LLMChatStreamItem} per SSE
   *  frame in Anthropic-style: provider_chosen first, then
   *  content_block_start / content_block_delta / content_block_stop
   *  pairs, then message_delta + done.
   *
   *  Iteration terminates when the gateway closes the stream. On a
   *  terminal error the gateway emits an `error` frame; the iterator
   *  yields it and the caller decides whether to throw.
   *
   *  Use this for live token streaming into LangChain BaseChatModel
   *  `_stream` callbacks. */
  async *llmStream(opts: LLMChatOptions): AsyncIterable<LLMChatStreamItem> {
    const body = serializeLLMOptions(opts, true);
    const resp = await this.ctx.fetchImpl(this.ctx.baseUrl + "/v1/_llm/chat", {
      method: "POST",
      headers: {
        ...authHeaders(this.ctx),
        "Content-Type": "application/json",
        Accept: "text/event-stream",
      },
      body: JSON.stringify(body),
      signal: opts.signal,
    });
    if (!resp.ok) {
      await raiseFromResponse(resp);
    }
    if (!resp.body) {
      throw new Error("llmStream: response has no body");
    }
    const reader = resp.body.getReader();
    yield* parseLLMStreamFrames(reader);
  }

  // ---- v0.11.4 OpenAI Embeddings compatibility shim ----

  /** Compute embedding vectors via the OpenAI-compatible
   *  `/v1/embeddings` endpoint. Dispatches to the single configured
   *  `providers.Embedder` (the same instance Memory tool uses for
   *  `embed:true` internally). No resolver path, no tier overlay,
   *  no streaming — embeddings are synchronous.
   *
   *  Accepts `input` as a string OR string array. Tokenized inputs
   *  (number arrays) are refused at the server boundary — send text.
   *  `encoding_format: "base64"` packs each float32 little-endian
   *  then base64-encodes (saves ~25% wire bytes on 1536-dim
   *  vectors); "float" (default) emits each vector as a JSON array.
   *
   *  Raises {@link AuthError} on 401; {@link UnavailableError} on
   *  503 (no embedder configured — set memory.embedder.* in
   *  loomcycle.yaml); {@link InvalidArgumentError} on 400 (bad
   *  request shape). */
  async embeddings(opts: LLMEmbeddingsOptions): Promise<LLMEmbeddingsResponse> {
    const { signal: _signal, ...body } = opts;
    return postJSON<LLMEmbeddingsResponse>(this.ctx, "/v1/embeddings", body, {
      signal: opts.signal,
    });
  }

  // ---- Internal helpers ----

  /** Shared SSE POST → stream-of-AgentEvent path. Used by
   *  runStreaming + continueSession.
   *
   *  When `debug` is true, the iterator yields a synthetic
   *  `{ type: "_meta", meta_subtype: "stream_open" }` before any real
   *  events AND a `{ type: "_meta", meta_subtype: "stream_close",
   *  meta_reason }` on EOF / abort / error. The default is silent
   *  (matches pre-v0.9.x behaviour). */
  /** GET-based SSE stream (RFC AI re-attach). The interactive re-attach
   *  endpoint is a GET with a query string, not a POST body, so it can't reuse
   *  streamSSE; the parse + error handling are identical otherwise. No debug
   *  shape — re-attach is a plain tail. */
  private async *streamSSEGet(
    path: string,
    signal?: AbortSignal,
  ): AsyncIterable<AgentEvent> {
    const headers: Record<string, string> = {
      Accept: "text/event-stream, application/json",
    };
    if (this.ctx.authToken) headers.Authorization = `Bearer ${this.ctx.authToken}`;
    const resp = await this.ctx.fetchImpl(this.ctx.baseUrl + path, {
      method: "GET",
      headers,
      signal,
    });
    if (!resp.ok) {
      await raiseFromResponse(resp);
    }
    if (!resp.body) {
      throw new Error("loomcycle: response has no body");
    }
    yield* parseSSE(resp.body.getReader());
  }

  private async *streamSSE(
    path: string,
    body: Record<string, unknown>,
    signal?: AbortSignal,
    debug?: boolean,
  ): AsyncIterable<AgentEvent> {
    const headers: Record<string, string> = {
      "Content-Type": "application/json",
      // Accept BOTH text/event-stream (the success path) AND
      // application/json (the error path — non-2xx responses come
      // back as JSON so raiseFromResponse can extract typed errors).
      // Per the Streamable HTTP spec; strict reverse proxies in
      // front of the sidecar 406 otherwise. Same rationale as the
      // v0.8.x MCP HTTP-transport hardening note in CLAUDE.md.
      Accept: "text/event-stream, application/json",
    };
    if (this.ctx.authToken) headers.Authorization = `Bearer ${this.ctx.authToken}`;

    const resp = await this.ctx.fetchImpl(this.ctx.baseUrl + path, {
      method: "POST",
      headers,
      body: JSON.stringify(body),
      signal,
    });

    if (!resp.ok) {
      await raiseFromResponse(resp);
    }
    if (!resp.body) {
      throw new Error("loomcycle: response has no body");
    }

    if (!debug) {
      // Silent default — pre-v0.9.x shape.
      yield* parseSSE(resp.body.getReader());
      return;
    }

    // Debug shape: synthetic open + close around the real stream.
    // The open frame carries no meta_reason — the frame itself IS the
    // signal. The close frame's meta_reason distinguishes normal EOF
    // from caller-side abort or a typed-error throw mid-stream.
    //
    // Close is emitted on both paths via try/catch/throw: success path
    // emits AFTER the try block; error path emits INSIDE the catch
    // before re-throwing. NOT a try/finally — the duplication is
    // intentional so the close-then-throw ordering is explicit and
    // a refactor adding `finally` doesn't accidentally double-emit.
    yield { type: "_meta", meta_subtype: "stream_open" };
    let closeReason = "eof";
    try {
      yield* parseSSE(resp.body.getReader());
    } catch (e) {
      // Capture the error type for the close frame, then re-throw so
      // typed-error handling at the consumer site still works.
      closeReason =
        e && typeof e === "object" && "name" in e
          ? String((e as { name: string }).name)
          : "error";
      yield {
        type: "_meta",
        meta_subtype: "stream_close",
        meta_reason: closeReason,
      };
      throw e;
    }
    yield {
      type: "_meta",
      meta_subtype: "stream_close",
      meta_reason: closeReason,
    };
  }

  // ---- v0.9.x n8n RFC Phase 0 ----

  /** List every operator-declared channel with aggregate stats
   *  (message_count, oldest_visible_at, newest_visible_at).
   *  Channels with no published messages still appear with
   *  message_count=0. Orphaned message rows for un-declared channels
   *  also appear (forensic visibility). Mirrors GET /v1/_channels. */
  async listChannels(opts?: {
    signal?: AbortSignal;
  }): Promise<ListChannelsResponse> {
    return await jsonFetch<ListChannelsResponse>(
      this.ctx,
      "/v1/_channels",
      opts,
    );
  }

  // ---- v0.9.x Channel CRUD ----
  //
  // Four bearer-authed ops mirroring the in-band Channel tool's
  // publish/subscribe/peek/ack. Two URL families behind the
  // `scope` field:
  //   - scope: "global" → POST /v1/_channels/{name}/{op} (admin)
  //   - scope: "user"   → POST /v1/users/{userId}/channels/{name}/{op}
  //
  // The same operator bearer token guards both surfaces; the per-user
  // URL embeds the user_id in the path so a caller can't forge a
  // different user_id by lying in the body.
  //
  // Subscribe is a SINGLE-ROUND-TRIP long-poll, not an open stream.
  // For continuous delivery, call `subscribeChannel` in a loop (the
  // n8n trigger node's pattern). Auto-commits the cursor on non-empty
  // batches (at-most-once shape) — use `peekChannel` + explicit
  // `ackChannel` for at-least-once semantics.

  /** Publish a JSON payload to an operator-declared channel. Mirrors
   *  the in-band Channel tool's publish op semantics — including
   *  deferred delivery via `deliverAt` (RFC3339Nano).
   *
   *  Errors:
   *  - {@link NotFoundError} (404) when the channel isn't in operator
   *    yaml. The wire `code` is `channel_not_declared`.
   *  - {@link InvalidArgumentError} (400) on invalid scope / payload.
   *  - {@link AuthError} (401) on bearer mismatch. */
  async publishChannel(
    channel: string,
    opts: PublishChannelOptions,
  ): Promise<ChannelPublishResult> {
    const path = channelOpPath(channel, opts.scope, opts.userId, "publish");
    const body: Record<string, unknown> = { payload: opts.payload };
    if (opts.deliverAt) body.deliver_at = opts.deliverAt;
    return postJSON<ChannelPublishResult>(this.ctx, path, body, {
      signal: opts.signal,
    });
  }

  /** Read the next batch of messages from a channel. Single-round-
   *  trip long-poll: returns immediately if messages are present,
   *  otherwise waits up to `waitMs` for a publish. AUTO-COMMITS the
   *  cursor on a non-empty batch.
   *
   *  For at-least-once delivery (crash safety between "loomcycle
   *  returned the batch" and "consumer finished processing"), use
   *  {@link LoomcycleClient.peekChannel} + an explicit
   *  {@link LoomcycleClient.ackChannel} after durable processing. */
  async subscribeChannel(
    channel: string,
    opts: SubscribeChannelOptions,
  ): Promise<ChannelSubscribeResult> {
    const path = channelOpPath(channel, opts.scope, opts.userId, "subscribe");
    const body: Record<string, unknown> = {};
    if (opts.fromCursor !== undefined) body.from_cursor = opts.fromCursor;
    if (opts.maxMessages !== undefined) body.max_messages = opts.maxMessages;
    if (opts.waitMs !== undefined) body.wait_ms = opts.waitMs;
    return postJSON<ChannelSubscribeResult>(this.ctx, path, body, {
      signal: opts.signal,
    });
  }

  /** Non-destructive read — never advances the committed cursor.
   *  Use for at-least-once consumption patterns: peek, process the
   *  batch durably, then `ackChannel` to advance. Multiple consumers
   *  can peek the same channel without disturbing each other. */
  async peekChannel(
    channel: string,
    opts: PeekChannelOptions,
  ): Promise<ChannelPeekResult> {
    let path = channelOpPath(channel, opts.scope, opts.userId, "peek");
    const params: string[] = [];
    if (opts.fromCursor) params.push(`from_cursor=${encodeURIComponent(opts.fromCursor)}`);
    if (opts.maxMessages) params.push(`max_messages=${opts.maxMessages}`);
    if (params.length > 0) path += `?${params.join("&")}`;
    return jsonFetch<ChannelPeekResult>(this.ctx, path, { signal: opts.signal });
  }

  /** Advance the committed cursor for a (channel, scope, scope_id)
   *  tuple. Cursor must be monotonically forward — older cursors
   *  raise a {@link ConflictError} (HTTP 409, code
   *  `channel_cursor_regression`). */
  async ackChannel(
    channel: string,
    opts: AckChannelOptions,
  ): Promise<ChannelAckResult> {
    const path = channelOpPath(channel, opts.scope, opts.userId, "ack");
    return postJSON<ChannelAckResult>(
      this.ctx,
      path,
      { cursor: opts.cursor },
      { signal: opts.signal },
    );
  }

  // ---- RFC S client twins: fan-in (await) / fan-out (broadcast) ----
  //
  // Multi-channel meta-ops (POST /v1/_channels/_await and /_broadcast).
  // Unlike the per-channel ops above they aren't addressed by name — the
  // whole request is the body — so they don't use channelOpPath. `scope`
  // + `userId` apply to every channel in the set. The complements to the
  // in-band Channel.await / Channel.broadcast tool ops, letting a wire
  // caller join independent producers / ping N workers over the same bus.

  /** Fan IN across `channels`: wait until the `mode` predicate is met
   *  (`any` / `all` / `at_least` N), or `waitMs` elapses. NON-committing
   *  (detection only) — `subscribe`/`ack` exactly what you process. A
   *  timeout is not an error: the result carries `timed_out: true`. */
  async awaitChannels(opts: AwaitChannelsOptions): Promise<ChannelAwaitResult> {
    const body: Record<string, unknown> = { channels: opts.channels };
    if (opts.scope) body.scope = opts.scope;
    if (opts.userId) body.scope_id = opts.userId;
    if (opts.mode) body.mode = opts.mode;
    if (opts.n !== undefined) body.n = opts.n;
    if (opts.fromCursor !== undefined) body.from_cursor = opts.fromCursor;
    if (opts.maxMessages !== undefined) body.max_messages = opts.maxMessages;
    if (opts.waitMs !== undefined) body.wait_ms = opts.waitMs;
    return postJSON<ChannelAwaitResult>(this.ctx, "/v1/_channels/_await", body, {
      signal: opts.signal,
    });
  }

  /** Fan OUT: publish the same `payload` to every channel in `channels`,
   *  in one call. Atomic at the declare pre-flight — one undeclared
   *  channel rejects the whole call ({@link NotFoundError}, code
   *  `channel_not_declared`) with nothing published. A per-channel write
   *  fault after that is reported in that channel's `results` entry while
   *  the successful publishes stand. */
  async broadcastChannels(
    opts: BroadcastChannelsOptions,
  ): Promise<ChannelBroadcastResult> {
    const body: Record<string, unknown> = {
      channels: opts.channels,
      payload: opts.payload,
    };
    if (opts.scope) body.scope = opts.scope;
    if (opts.userId) body.scope_id = opts.userId;
    if (opts.deliverAt) body.deliver_at = opts.deliverAt;
    return postJSON<ChannelBroadcastResult>(
      this.ctx,
      "/v1/_channels/_broadcast",
      body,
      { signal: opts.signal },
    );
  }

  // ---- v0.11.5 Channel admin CRUD ----
  //
  // Three bearer-authed ops that mutate the runtime-substrate
  // channel registry. yaml-declared channels are immutable from
  // this surface — the server refuses with HTTP 409 + wire `code`
  // `channel_yaml_immutable`. The TS adapter surfaces that as a
  // {@link LoomcycleError} with status 409.

  /** Create a new runtime-substrate channel. Refuses with HTTP 409
   *  when the name matches a yaml-declared channel (code
   *  `channel_yaml_immutable`) or an existing runtime channel
   *  (code `channel_name_in_use`). */
  async createChannel(opts: CreateChannelOptions): Promise<ChannelDescriptor> {
    const body: Record<string, unknown> = { name: opts.name };
    if (opts.description !== undefined) body.description = opts.description;
    if (opts.scope !== undefined) body.scope = opts.scope;
    if (opts.semantic !== undefined) body.semantic = opts.semantic;
    if (opts.default_ttl !== undefined) body.default_ttl = opts.default_ttl;
    if (opts.max_messages !== undefined) body.max_messages = opts.max_messages;
    if (opts.publisher !== undefined) body.publisher = opts.publisher;
    if (opts.period !== undefined) body.period = opts.period;
    return postJSON<ChannelDescriptor>(this.ctx, "/v1/_channels", body, {
      signal: opts.signal,
    });
  }

  /** Partially update a runtime-substrate channel. Nil-valued fields
   *  in `opts` leave the corresponding attribute unchanged. Refuses
   *  yaml-declared channels with HTTP 409. */
  async updateChannel(
    name: string,
    opts: UpdateChannelOptions,
  ): Promise<ChannelDescriptor> {
    const body: Record<string, unknown> = {};
    if (opts.description !== undefined) body.description = opts.description;
    if (opts.default_ttl !== undefined) body.default_ttl = opts.default_ttl;
    if (opts.max_messages !== undefined) body.max_messages = opts.max_messages;
    if (opts.semantic !== undefined) body.semantic = opts.semantic;
    return patchJSON<ChannelDescriptor>(
      this.ctx,
      `/v1/_channels/${encodeURIComponent(name)}`,
      body,
      { signal: opts.signal },
    );
  }

  /** Delete a runtime-substrate channel + cascade its persisted
   *  messages + cursors. yaml-declared channels refuse with HTTP 409.
   *  Idempotent: deleting a non-existent runtime channel returns a
   *  {@link NotFoundError} so the caller can distinguish that case. */
  async deleteChannel(
    name: string,
    opts?: { signal?: AbortSignal },
  ): Promise<void> {
    return deleteRequest(
      this.ctx,
      `/v1/_channels/${encodeURIComponent(name)}`,
      opts,
    );
  }

  /** Clear all buffered messages on a channel WITHOUT removing its
   *  definition or subscriber cursors. Unlike {@link deleteChannel} this
   *  is allowed on yaml-declared channels too — draining a yaml channel
   *  that filled with test traffic was the F20 pain it solves. Returns
   *  the channel name + the number of messages cleared. Unknown channels
   *  reject with a {@link NotFoundError}. */
  async purgeChannel(
    name: string,
    opts?: { signal?: AbortSignal },
  ): Promise<ChannelPurgeResult> {
    return postJSON<ChannelPurgeResult>(
      this.ctx,
      `/v1/_channels/${encodeURIComponent(name)}/purge`,
      {},
      { signal: opts?.signal },
    );
  }

  // ---- v0.11.5 Memory entry admin CRUD ----

  /** Idempotently upsert one memory entry by full (scope, scope_id,
   *  key) identifier. PUT semantics — re-writes overwrite the value.
   *  Optional embed flag triggers a synchronous embed via the
   *  operator-configured embedder. */
  async setMemoryEntry(
    scope: string,
    scopeID: string,
    key: string,
    opts: SetMemoryEntryOptions,
  ): Promise<SetMemoryEntryResponse> {
    const body: Record<string, unknown> = { value: opts.value };
    if (opts.embed !== undefined) body.embed = opts.embed;
    if (opts.ttl_seconds !== undefined) body.ttl_seconds = opts.ttl_seconds;
    return putJSON<SetMemoryEntryResponse>(
      this.ctx,
      `/v1/_memory/scopes/${encodeURIComponent(scope)}/${encodeURIComponent(scopeID)}/keys/${encodeURIComponent(key)}`,
      body,
      { signal: opts.signal },
    );
  }

  /** Delete one memory entry by (scope, scope_id, key). Idempotent:
   *  deleting a missing row is a non-error per the in-band Memory
   *  tool's semantics — both surfaces return 204. */
  async deleteMemoryEntry(
    scope: string,
    scopeID: string,
    key: string,
    opts?: { signal?: AbortSignal },
  ): Promise<void> {
    return deleteRequest(
      this.ctx,
      `/v1/_memory/scopes/${encodeURIComponent(scope)}/${encodeURIComponent(scopeID)}/keys/${encodeURIComponent(key)}`,
      opts,
    );
  }

  /** Subscribe to run state transitions for one user_id via SSE.
   *  Yields one `{ kind: "open", ... }` item first (confirms the
   *  connection is live), then one `{ kind: "event", ... }` per
   *  matching state transition until the stream closes.
   *
   *  The stream stays open for at most 30 minutes (server-enforced).
   *  Callers running indefinitely should reconnect on close.
   *
   *  Errors during the stream throw — they do NOT surface as items.
   *  Pass an AbortSignal to terminate cleanly from the consumer side.
   *
   *  v0.9.x options:
   *  - `parentAgentId` — client-side filter: only `kind: "event"`
   *    items whose payload's `parent_agent_id` matches are yielded.
   *    The server still streams every matching event; the adapter
   *    filters before yielding. Empty/omitted = no filter.
   *  - `debug` — when true, an additional `{ kind: "close", payload:
   *    { reason } }` item is yielded when the stream ends (EOF,
   *    abort, or pre-yield error). Useful for n8n nodes that surface
   *    "stream re-opened / closed" log entries without inferring
   *    from timing. Default false. */
  async *streamUserRunStates(
    userId: string,
    opts?: StreamUserRunStatesOptions,
  ): AsyncIterable<RunStateStreamItem> {
    const params = new URLSearchParams();
    if (opts?.statuses && opts.statuses.length > 0) {
      params.set("status", opts.statuses.join(","));
    }
    if (opts?.agent) {
      params.set("agent", opts.agent);
    }
    const qs = params.toString();
    const path =
      `/v1/users/${encodeURIComponent(userId)}/agents/stream` +
      (qs ? `?${qs}` : "");

    const headers: Record<string, string> = {
      Accept: "text/event-stream",
    };
    if (this.ctx.authToken) {
      headers.Authorization = `Bearer ${this.ctx.authToken}`;
    }

    const resp = await this.ctx.fetchImpl(this.ctx.baseUrl + path, {
      method: "GET",
      headers,
      signal: opts?.signal,
    });

    if (!resp.ok) {
      await raiseFromResponse(resp);
    }
    if (!resp.body) {
      throw new Error("loomcycle: streamUserRunStates response has no body");
    }

    const parentFilter = opts?.parentAgentId ?? "";
    const debug = opts?.debug === true;

    let closeReason = "eof";
    try {
      for await (const item of parseRunStateSSE(resp.body.getReader())) {
        // Client-side parent_agent_id filter. Pre-v1 the server has no
        // ?parent_agent_id= query param; n8n-style consumers that need
        // a narrow view get a smaller iterator at the cost of
        // unchanged server load. See StreamUserRunStatesOptions for
        // the trade-off note.
        if (
          parentFilter !== "" &&
          item.kind === "event" &&
          item.payload.parent_agent_id !== parentFilter
        ) {
          continue;
        }
        yield item;
      }
    } catch (e) {
      closeReason =
        e && typeof e === "object" && "name" in e
          ? String((e as { name: string }).name)
          : "error";
      if (debug) {
        yield { kind: "close", payload: { reason: closeReason } };
      }
      throw e;
    }
    if (debug) {
      yield { kind: "close", payload: { reason: closeReason } };
    }
  }
}

/** Lightweight SSE parser tailored to the run-state stream. Each
 *  frame's event name distinguishes the two kinds; data is JSON.
 *  Comment lines (": keepalive") are ignored. */
async function* parseRunStateSSE(
  reader: ReadableStreamDefaultReader<Uint8Array>,
): AsyncIterable<RunStateStreamItem> {
  const decoder = new TextDecoder("utf-8");
  let buf = "";
  let event = "";
  let data = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });

    let idx;
    while ((idx = buf.indexOf("\n")) !== -1) {
      const line = buf.slice(0, idx).replace(/\r$/, "");
      buf = buf.slice(idx + 1);

      if (line === "") {
        if (event && data) {
          try {
            const parsed = JSON.parse(data) as unknown;
            if (event === "stream_open") {
              yield {
                kind: "open",
                payload: parsed as RunStateStreamItem extends { kind: "open" }
                  ? RunStateStreamItem["payload"]
                  : never,
              } as RunStateStreamItem;
            } else if (event === "run_state") {
              yield {
                kind: "event",
                payload: parsed as RunStateStreamItem extends { kind: "event" }
                  ? RunStateStreamItem["payload"]
                  : never,
              } as RunStateStreamItem;
            }
          } catch {
            // Drop malformed frame silently — same posture as parseSSE.
          }
        }
        event = "";
        data = "";
        continue;
      }
      if (line.startsWith("event:")) event = line.slice("event:".length).trim();
      else if (line.startsWith("data:")) data = line.slice("data:".length).trim();
    }
  }
}

// channelOpPath builds the v0.9.x Channel CRUD URL. Two families:
//   - scope === "global" → /v1/_channels/{channel}/{op}
//   - scope === "user"   → /v1/users/{userId}/channels/{channel}/{op}
// Channel name is URL-encoded so names containing slashes
// ("findings/alpha", "_system/foo") survive transport.
function channelOpPath(
  channel: string,
  scope: "global" | "user",
  userId: string | undefined,
  op: "publish" | "subscribe" | "peek" | "ack",
): string {
  const enc = encodeURIComponent(channel);
  if (scope === "user") {
    if (!userId) {
      throw new Error(
        `loomcycle: scope="user" requires opts.userId for the channel ${op} call`,
      );
    }
    return `/v1/users/${encodeURIComponent(userId)}/channels/${enc}/${op}`;
  }
  return `/v1/_channels/${enc}/${op}`;
}

// browseQuery maps the RFC AS browse-by-subject opts to the wire query-param
// names the off-run Path/Document endpoints read from the URL (scope_id /
// tenant — matching web/src/api.ts:substratePost). Only set values are
// included; postJSON drops the query entirely when the map is empty, so a
// caller passing neither builds the same URL as before RFC AS.
function browseQuery(opts?: {
  scopeId?: string;
  tenant?: string;
}): Record<string, string> {
  const q: Record<string, string> = {};
  if (opts?.scopeId) q.scope_id = opts.scopeId;
  if (opts?.tenant) q.tenant = opts.tenant;
  return q;
}


// ---- v0.11.0 LLM Gateway helpers ----

/** serializeLLMOptions strips the AbortSignal (transport concern) and
 *  forces the stream flag to match the call mode. */
function serializeLLMOptions(opts: LLMChatOptions, stream: boolean): Record<string, unknown> {
  const { signal: _signal, ...rest } = opts;
  return { ...rest, stream };
}

/** parseLLMStreamFrames drains an SSE stream from /v1/_llm/chat and
 *  yields one LLMChatStreamItem per frame. Unlike parseSSE in
 *  stream.ts, the gateway's frames discriminate purely by the SSE
 *  event name — there's no `type` field on the data payload. */
async function* parseLLMStreamFrames(
  reader: ReadableStreamDefaultReader<Uint8Array>,
): AsyncIterable<LLMChatStreamItem> {
  const decoder = new TextDecoder("utf-8");
  let buf = "";
  let event = "";
  let data = "";

  const flush = (): LLMChatStreamItem | null => {
    if (!event || !data) {
      event = "";
      data = "";
      return null;
    }
    try {
      const payload = JSON.parse(data);
      const item = { kind: event, payload } as LLMChatStreamItem;
      event = "";
      data = "";
      return item;
    } catch {
      event = "";
      data = "";
      return null;
    }
  };

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });

    let idx;
    while ((idx = buf.indexOf("\n")) !== -1) {
      const line = buf.slice(0, idx).replace(/\r$/, "");
      buf = buf.slice(idx + 1);

      if (line === "") {
        const item = flush();
        if (item) yield item;
        continue;
      }
      if (line.startsWith("event:")) event = line.slice("event:".length).trim();
      else if (line.startsWith("data:")) data = line.slice("data:".length).trim();
      // Keepalive comment lines (`:keepalive`) are silently dropped.
    }
  }
  // Drain a final un-newlined frame on connection close.
  if (buf.length > 0) {
    const line = buf.replace(/\r$/, "");
    if (line.startsWith("event:")) event = line.slice("event:".length).trim();
    else if (line.startsWith("data:")) data = line.slice("data:".length).trim();
  }
  const item = flush();
  if (item) yield item;
}
