flowchart TD
    App[Application / UI] -->|prompt, steer, abort, config| Harness
    Harness -->|snapshot + events| App
    Harness -->|hooks + events| Ext[Extensions]
    Harness --> Loop[Step primitives
request / tools]
    Loop --> Provider[LLM provider]
    Loop --> Tools[Tools]
    Harness --> Session
    Session --> Storage[(JSONL / memory / SQLite)]
    Harness -.->|telemetry| Otel[Observability]

stateDiagram-v2
    [*] --> Idle
    Idle --> Running: prompt accepted
    Running --> Idle: finished
    Running --> Cancelling: abort
    Cancelling --> Idle: reconciled
    Running --> Faulted: append failure
    Suspended --> Running: resume
    Suspended --> Cancelling: abort

flowchart TD
    CP[Checkpoint] --> DW[Flush deferred writes]
    DW --> Q[Consume queued messages]
    Q --> AC{Context too big?}
    AC -->|yes| C[Auto-compact] --> S
    AC -->|no| S[Step]
    S --> More{Tool calls or
queued messages?}
    More -->|yes| CP
    More -->|no| End[before_run_end hook]
    End -->|returned follow-up| CP
    End -->|nothing| F[Finish run]

flowchart LR
    R[Request] -->|retry| R
    R --> A[Assistant response persisted]
    A --> T[Tool batch] --> E[Step end]

flowchart TB
    subgraph LOG ["log (append order, top to bottom)"]
        direction TB
        h1(["op_started op-1"])
        u1["message user U1"]
        h2(["generation_started"])
        a1["message assistant A1"]
        h3(["tool_started call-1"])
        h4(["queue_enqueued steer S1"])
        t1["message toolResult T1"]
        s1["message user S1"]
        a2["message assistant A2"]
        h5(["op_finished op-1"])
        h1 ~~~ u1 ~~~ h2 ~~~ a1 ~~~ h3 ~~~ h4 ~~~ t1 ~~~ s1 ~~~ a2 ~~~ h5
    end
    u1 -->|parent| a1 -->|parent| t1 -->|parent| s1 -->|parent| a2

/** A session entry payload with its id pre-allocated. parentId and timestamp
    are assigned when the entry is actually appended: it becomes a child of the
    then-current leaf, exactly like a normal append. */
type ProvisionedEntry<T extends SessionTreeEntry = SessionTreeEntry> = Omit<T, "parentId" | "timestamp">;
type ProvisionedMessage = ProvisionedEntry<MessageEntry>;

interface HarnessEntryBase {
  id: string;
  seq: number;            // position in the chronological log
  ref: string;            // the ref this record belongs to ("main" in a single-ref session)
  timestamp: string;
}
// Session entries carry no ref — the tree is shared between refs (common
// prefixes), so a ref field would fake ownership that does not exist. Which
// ref appended a session entry is derivable: a ref's operation appends a
// chain from its anchor, so membership is parentId linkage into that chain —
// one pass over the bounded tail, in seq order. Leaf records are the
// exception: they carry ref explicitly, they ARE the per-ref pointer.
// Old files: everything reads as "main".
// Entries that belong to an operation carry runId: the id of that operation's
// operation_started entry. Not on the base: queue_enqueued(nextRun) belongs to
// no operation — it targets the next run, and can be accepted while idle.

// The durable acceptance boundary for an operation. Everything decided
// before acceptance is persisted here: before_run output, queued next-run
// consumption, provisioned ids for structural results.
interface OperationStartedEntry extends HarnessEntryBase {
  type: "operation_started";
  sourceLeafId: string | null;      // the ref's leaf at acceptance
  intent:
    | {
        kind: "run";
        /** Prompt + before_run injections, full payloads with provisioned ids. */
        initialMessages: ProvisionedMessage[];
        /** Set iff before_run overrode the system prompt; fixed for the run.
            Absent: the systemPrompt config callback is evaluated per request
            (sees current active tools — mid-run tool changes rebuild the prompt). */
        systemPromptOverride?: string;
        resumeData?: Record<string, JsonValue>;   // per extension id
      }
    | {
        kind: "compaction";
        customInstructions?: string;
        resultEntryId: string;
      }
    | {
        kind: "navigation";
        targetId: string;
        destinationLeafId: string | null;
        summarize: boolean;
        customInstructions?: string;
        label?: string;
        summaryEntryId?: string;
        labelEntryId?: string;
        leafEntryId: string;
      };
}

// Cancellation is durable the moment abort() resolves. Clears this
// operation's steer/follow-up queue items; next-run items survive.
interface OperationCancelledEntry extends HarnessEntryBase {
  type: "operation_cancelled";
  runId: string;
  reason: "user" | "shutdown";
}

// Closes the operation. Cursor-neutral: cannot undo a navigation's leaf move.
// outcome "failed" is a durable, orderly failure: retry attempts exhausted,
// compaction/summary generation permanently failing, required model missing.
// outcome "cancelled" is a structural operation declined by its hook
// (before_compaction/before_navigation cancel). Either way the operation is
// closed; nothing resumes. Distinct from Faulted (section 4): a fault means
// appends fail, so no finish entry can be written and the operation restores
// as suspended instead.
interface OperationFinishedEntry extends HarnessEntryBase {
  type: "operation_finished";
  runId: string;
  outcome: "completed" | "aborted" | "failed" | "cancelled";
  error?: { code: string; message: string };   // safe fields only, no payloads
}

// Appended before each generation cycle the harness bills to a provider — an
// uncertainty marker ("a request may have gone out and been billed") and the
// durable crash-loop bound: attempt counts survive restarts. One entry per
// cycle, even when a cycle makes several physical requests (split-turn
// compaction runs two); recovery granularity is the result entry, so finer
// accounting buys nothing. Position and completion are positional: attempts
// after the newest session entry belong to the current request cycle, and a
// session entry committing closes the cycle. Validation: attempt numbers are
// consecutive within each gap between session entries (a gap is always
// single-purpose — compaction either commits its entry, closing the gap,
// or fails the run). All positional rules are per ref: "newest session
// entry" means the newest session entry chained by this ref's operation
// (parentId membership, see above) — the per-ref partition is what makes
// positional reduction safe under interleaved refs.
interface GenerationStartedEntry extends HarnessEntryBase {
  type: "generation_started";
  runId: string;
  purpose: "step" | "compaction" | "branch_summary";
  attempt: number;                  // 1-based within the current cycle
  model: { provider: string; modelId: string };
}

// Appended after before_tool and validation, before the effect starts.
// assistantEntryId + toolIndex is the durable invocation identity.
interface ToolStartedEntry extends HarnessEntryBase {
  type: "tool_started";
  runId: string;
  assistantEntryId: string;
  toolIndex: number;
  toolCallId: string;
  toolName: string;
  effectiveArgs: Record<string, unknown>;   // post-before_tool
  resultEntryId: string;                    // provisioned
  replay: "never" | "safe";
}

// steer()/followUp()/nextRun() acceptance. The message payload (any
// AgentMessage that converts to a user LLM message) travels here; the session
// entry appears at the consumption point. Steer/follow-up items resolve within
// their run. Next-run items are consumed by the next run-kind operation on the
// same ref, embedded in its operation_started initialMessages; compaction and
// navigation pass through without consuming. Pending items therefore always
// sit after the ref's last run-kind operation_started — recovery never scans
// further back.
interface QueueEnqueuedEntry extends HarnessEntryBase {
  type: "queue_enqueued";
  queue: "steer" | "followUp" | "nextRun";
  /** steer/followUp: their active run. Absent for nextRun — the item belongs
      to no operation until the next run consumes it. */
  runId?: string;
  target: ProvisionedMessage;
}

// SessionTree write or config setter accepted mid-step: message, custom entry,
// label, session name, or config change (model_change, ...). Applied in
// acceptance order at the next checkpoint — live or during recovery, each
// target is appended as a child of the then-current leaf; that is why
// ProvisionedEntry omits parentId.
interface WriteDeferredEntry extends HarnessEntryBase {
  type: "write_deferred";
  runId: string;
  target: ProvisionedEntry;   // full payload, provisioned id
}

/** Filters and paging. Omit type to match every entry. */
interface EntryQuery {
  type?: SessionTreeEntry["type"];
  customType?: string;                     // for type: "custom"
  order?: "newestFirst" | "oldestFirst";   // default newestFirst
  limit?: number;
  cursor?: EntryCursor;                    // continue a previous page
}

/** Where a branch scan starts and stops. Defaults: the whole path, leaf to root. */
interface BranchBounds {
  /** Leaf end of the path. Default: the view's ref leaf. Needed to query another branch or an old compaction tail. */
  start?: string;
  /** Scan ends after the first matching entry, inclusive. */
  stopAtType?: SessionTreeEntry["type"];
  stopAtId?: string;
}

// Generic over metadata; finders return Extract<SessionTreeEntry, { type: T }>.
interface SessionTree<TMetadata extends SessionMetadata = SessionMetadata> {
  // Reads — always allowed
  getMetadata(): Promise<TMetadata>;
  getEntry(id: string): Promise<SessionTreeEntry | undefined>;
  getLeafId(): Promise<string | null>;
  getStats(): Promise<SessionStats>;

  // Global facts — latest-wins records outside the tree, not branch-scoped.
  // Setter naming is deliberate: "append" is reserved for tree writes.
  getName(): Promise<string | undefined>;
  setName(name: string): Promise<string>;
  getLabel(id: string): Promise<string | undefined>;
  setLabel(targetId: string, label: string | undefined): Promise<string>;

  /** Session-wide: all branches, log order. */
  findEntries(query?: EntryQuery): Promise<SessionTreeEntry[]>;

  /** Sugar: findEntries with limit 1. */
  findEntry(query?: Omit<EntryQuery, "limit" | "cursor">): Promise<SessionTreeEntry | undefined>;

  /** Branch-scoped: the path start (default leaf) → root. */
  findEntriesOnBranch(query?: EntryQuery & BranchBounds): Promise<SessionTreeEntry[]>;

  /** Sugar: findEntriesOnBranch with limit 1. */
  findEntryOnBranch(query?: Omit<EntryQuery, "limit" | "cursor"> & BranchBounds): Promise<SessionTreeEntry | undefined>;

  // Writes — immediate on standalone Session, deferred while a harness runs.
  // Resolve on durable acceptance; the returned string is the provisioned entry
  // id the session entry will carry when applied. Safe to call from hook and
  // event handlers at any point.
  appendMessage(message: AgentMessage): Promise<string>;   // includes custom app message types
  appendCustomEntry(customType: string, data?: unknown): Promise<string>;

  /** Writes accepted but not yet applied, in acceptance order. Empty on standalone Session. */
  getPendingWrites(): { id: string; entry: SessionTreeEntry }[];
}

class Session<TMetadata> implements SessionTree<TMetadata> {
  appendHarnessEntry(entry: HarnessEntryInput): Promise<HarnessEntry>;
  /** Full chronological log — session and harness entries interleaved. */
  getLog(options?: { afterSeq?: number; limit?: number }): Promise<LogRecord[]>;

  /** Typed harness-entry queries, same shape as the tree finders. SQLite
      serves them from an indexed harness-entry table. */
  findHarnessEntries(query?: { type?: HarnessEntry["type"]; ref?: string; runId?: string; afterSeq?: number; order?: "newestFirst" | "oldestFirst"; limit?: number }): Promise<HarnessEntry[]>;
  findHarnessEntry(query?): Promise<HarnessEntry | undefined>;   // limit 1
}

tree (shared, append-only)              refs
a ── b ── c ── d                        main → d
      └── e ── f                        slack:1719432.0021 → f

const { harness, suspended } = await AgentHarness.create({ session, models, model, ... });

for (const s of suspended) {
  await harness.ref(s.ref)!.resume();   // or .abort(); interactive pi: 0 or 1, always "main"
}
await harness.prompt("...");

interface AgentHarnessOptions {
  session: Session;
  /** Provider collection for all requests: steps, compaction, branch summaries. */
  models: Models;
  /** model, thinkingLevel, activeToolNames: initial values only. If a ref's
      branch has persisted config entries, the session wins; these apply to fresh
      sessions and refs without config history. An unresolvable persisted model
      surfaces in suspended[].missing / falls back with a warning, mirroring the
      old coding agent. */
  model: Model<any>;
  thinkingLevel?: ThinkingLevel;
  tools?: TTool[];
  activeToolNames?: string[];
  toolContext?: TContext | (() => TContext | Promise<TContext>);
  systemPrompt?: AgentHarnessSystemPrompt;
  resources?: AgentHarnessResources;
  /** Curated provider request options (transport, headers, timeouts, provider-internal retries). Snapshotted at request start. */
  streamOptions?: AgentHarnessStreamOptions;
  /** Harness-level retry for failed requests (steps, compaction, branch summaries). Attempt counts are durable; a restart never resets them. */
  retry?: RetryPolicy;
  compaction?: CompactionSettings;   // enabled, reserveTokens, keepRecentTokens
  steeringMode?: QueueMode;
  followUpMode?: QueueMode;
  /** Converts AgentMessages (including app custom types via CustomAgentMessages)
      to provider messages before each request. Default: exported
      defaultConvertToLlm, handling bashExecution, custom, branchSummary,
      compactionSummary; standard messages pass through. Also used at
      prompt/queue acceptance to validate that a submitted AgentMessage
      converts to a user message. Reordering/pruning is not its job; that is
      transform_context. */
  convertToLlm?: (messages: AgentMessage[]) => Message[] | Promise<Message[]>;
}

/** The operation surface of one ref. AgentHarness implements it for main;
    harness.ref(name) returns the same surface for any other ref. */
interface AgentRef {
  readonly name: string;              // "main" on the harness itself
  getLeafId(): Promise<string | null>;

  // Operations. Never throw — they resolve with a result value in every case
  // (see "Results, not exceptions" below). Message forms mirror Agent: any
  // AgentMessage accepted here (and by the queues) must convert to a user LLM
  // message via convertToLlm; validated at acceptance. At most one operation
  // is active per ref; operations on different refs run concurrently.
  prompt(text: string, images?: ImageContent[]): Promise<RunResult>;
  prompt(message: AgentMessage | AgentMessage[]): Promise<RunResult>;   // e.g. sendMessage triggerTurn
  skill(name: string, additionalInstructions?: string): Promise<RunResult>;
  promptFromTemplate(name: string, args?: string[]): Promise<RunResult>;
  compact(options?: { customInstructions?: string; settings?: Partial<CompactionSettings> }): Promise<CompactionRunResult>;
  navigateTree(targetId: string, options?: NavigateTreeOptions): Promise<NavigationRunResult>;

  /** New. Continue this ref's suspended operation to its durable end. */
  resume(): Promise<ResumeResult>;

  /** Existing. Now durable: cancellation is recorded before effects are signalled; returns without waiting for reconciliation. No-op while this ref is idle. */
  abort(): Promise<AbortResult>;

  // Queues (existing) — now durable on resolve, and per ref (nextRun included:
  // consumed by this ref's next run). Payloads are AgentMessage: user messages
  // and custom extension messages (sendMessage deliverAs). As with prompt(),
  // every AgentMessage must convert to a user LLM message via convertToLlm;
  // validated at acceptance.
  steer(text: string, images?: ImageContent[]): Promise<QueueResult>;     // requires active run
  steer(message: AgentMessage): Promise<QueueResult>;
  followUp(text: string, images?: ImageContent[]): Promise<QueueResult>;  // requires active run
  followUp(message: AgentMessage): Promise<QueueResult>;
  nextRun(text: string, images?: ImageContent[]): Promise<QueueResult>;   // any time; was nextTurn()
  nextRun(message: AgentMessage): Promise<QueueResult>;

  // Idle coordination — this ref's idleness.
  waitForIdle(): Promise<void>;              // existing
  runWhenIdle(callback): Promise<void>;      // new; runtime-only, not durable

  // Persisted, branch-anchored config — per ref by construction: session
  // entries on the path behind this ref's leaf, restored by point queries.
  // Setters mid-step become deferred writes on this ref.
  getModel() / setModel(model)
  getThinkingLevel() / setThinkingLevel(level)
  getActiveTools() / setActiveTools(toolNames)    // unknown names reject

  /** This ref's SessionTree view (section 5): branch reads default to this
      ref's leaf, appends chain to it, writes defer while this ref runs.
      Replaces appendMessage(). */
  session: SessionTree;

  /** Scoped: this ref's transcript, run state, queues, and events (section 8). */
  watch(): Promise<{ snapshot: RefSnapshot, start, unsubscribe }>;
}

class AgentHarness implements AgentRef {
  /** Opens the session log, restores state, starts no effects. Replaces the
      constructor. One suspended entry per ref with an unfinished operation. */
  static create(options: AgentHarnessOptions): Promise<{
    harness: AgentHarness;
    suspended: SuspendedOperation[];
  }>;

  // Ref management. Names are app-chosen keys ("slack:1719432.0021").
  ref(name: string): AgentRef | undefined;             // lookup, never creates
  createRef(name: string, at: string | null): Promise<RefResult>;   // anchor at an entry or root
  deleteRef(name: string): Promise<RefResult>;         // pointer only; rejected while its
                                                       // operation is active or suspended;
                                                       // "main" cannot be deleted
  refs(): RefInfo[];

  // Harness-global config — registries and runtime capabilities, shared by
  // all refs. Tool implementations are code and cannot persist; the active
  // set (names) is what persists, per ref.
  getTools() / setTools(tools, activeToolNames?)  // registry; active set applies to main
  getResources() / setResources(resources)
  getStreamOptions() / setStreamOptions(streamOptions)
  getRetryPolicy() / setRetryPolicy(policy)            // new
  getCompactionSettings() / setCompactionSettings(s)   // new
  getSteeringMode() / setSteeringMode(mode)
  getFollowUpMode() / setFollowUpMode(mode)

  /** Session-wide observer: refs inventory snapshot plus the unfiltered event
      stream (section 9). No transcripts — compose with ref.watch() per ref. */
  watchSession(): Promise<{ snapshot: SessionSnapshot, start, unsubscribe }>;

  // Harness-global; every hook/event payload carries ref (sections 9, 10).
  hooks: ...;
  events: ...;

  /** New. Detach cleanly — see semantics below. Does not abort operations; they stay resumable. */
  close(): Promise<void>;
}

interface RefInfo {
  name: string;
  leafId: string | null;
  run: null | { id: string; kind: "run" | "compaction" | "navigation";
                status: "running" | "suspended" | "cancelling" };
}

type RefResult = { ok: true; ref: AgentRef } | { ok: false; outcome: "rejected"; error: ErrorInfo };

interface SuspendedOperation {
  ref: string;
  kind: "run" | "compaction" | "navigation";
  id: string;
  startedAt: string;
  /** For runs: original prompt content (text and images), for display. */
  prompt?: (TextContent | ImageContent)[];
  /** The operation was cancelled pre-crash; resume() completes the abort. Undelivered
      steer/follow-up payloads are returned here — the crash-path equivalent of
      AbortResult — so a client can offer to requeue them. */
  cancelled?: { clearedSteer: AgentMessage[]; clearedFollowUp: AgentMessage[] };
  /** Identities the log references that current config cannot resolve. Non-empty: resume() resolves rejected. */
  missing: { tools: string[]; models: string[] };
}

interface ErrorInfo { code: string; message: string }

/** Failures shared by all methods. */
type Failure =
  | { ok: false; outcome: "rejected"; error: ErrorInfo }
    // the call never became an operation — no runId exists anywhere, the log is untouched
  | { ok: false; outcome: "faulted"; runId?: string; error: ErrorInfo };
    // appends stopped working. runId present: the operation started and restores as
    // suspended after reopening. absent: the fault hit the acceptance append itself.

// finalMessage is defined via the entry→message projection: the run's newest
// message entry that projects to an AssistantMessage. Custom entries project to
// nothing and custom messages project to user messages, so neither can be
// finalMessage by construction. Full transcript content is not duplicated in
// results — it is in the session (branch query scoped to the run) and was
// delivered via events.
type RunResult =
  | { ok: true; runId: string; finalMessage: AssistantMessage }
  | { ok: false; outcome: "aborted"; runId: string; finalMessage: AssistantMessage }  // the aborted closure
  | { ok: false; outcome: "failed";  runId: string; error: ErrorInfo; finalMessage?: AssistantMessage }
    // finalMessage absent when the run failed before any assistant response (e.g. auto-compaction)
  | Failure;

type CompactionRunResult =
  | { ok: true; runId: string; entry: CompactionEntry }
  | { ok: false; outcome: "cancelled" | "aborted"; runId: string }  // hook declined / user abort
  | { ok: false; outcome: "failed"; runId: string; error: ErrorInfo }
  | Failure;

type NavigationRunResult =
  | { ok: true; runId: string; newLeafId: string | null; summaryEntry?: BranchSummaryEntry }
  | { ok: false; outcome: "cancelled" | "aborted"; runId: string }
  | { ok: false; outcome: "failed"; runId: string; error: ErrorInfo }
  | Failure;

type QueueResult =
  | { ok: true }           // durably accepted in the log
  | Failure;

// Runs can never be "cancelled" (no hook vetoes an accepted run);
// kind discriminates which result shape applies.
type ResumeResult =
  | ({ kind: "run" } & RunResult)
  | ({ kind: "compaction" } & CompactionRunResult)
  | ({ kind: "navigation" } & NavigationRunResult);

const result = await harness.prompt("...");
if (!result.ok) {
  showError(result);          // switch on result.outcome for finer handling
  return;
}
render(result.finalMessage);

const { snapshot, start, unsubscribe } = await ref.watch();   // harness.watch() = main's

await send(client, { kind: "snapshot", snapshot });      // snapshot is on the wire
start((event) => send(client, event));                   // flush buffer in order, then go live

interface RefSnapshot {
  ref: string;
  // Transcript: this ref's branch, oldest first (the context window plus
  // its compaction entry; UIs page further history via session queries)
  transcript: SessionTreeEntry[];
  leafId: string | null;

  run: null | {
    id: string;
    kind: "run" | "compaction" | "navigation";
    status: "running" | "suspended" | "cancelling";
    startedAt: string;
    /** status "suspended" only: what a client needs to offer resume/abort.
        Same data create() returned — duplicated here because a remote UI
        only ever sees the snapshot, not create()'s return value. */
    suspended?: {
      prompt?: (TextContent | ImageContent)[];
      missing: { tools: string[]; models: string[] };
      cancelled?: { clearedSteer: AgentMessage[]; clearedFollowUp: AgentMessage[] };
    };
    /** Live progress, when mid-step. */
    streamingMessage?: AssistantMessage;
    runningTools: {
      /** Unique within the current tool batch; correlates with the tool-call block in the newest assistant message. */
      toolCallId: string;
      toolName: string;
      args: unknown;
      /** Latest streamed partial result, when the tool reports updates. */
      partialResult?: AgentToolResult<unknown>;
    }[];
    retry?: { attempt: number; maxAttempts: number; nextAttemptAt: string };
  };

  queues: { steer: AgentMessage[]; followUp: AgentMessage[]; nextRun: AgentMessage[] };
  pendingWrites: { id: string; entry: SessionTreeEntry }[];

  faulted: boolean;   // harness-wide; mirrored into every ref snapshot
}

interface SessionSnapshot {
  refs: (RefInfo & {
    suspended?: SuspendedOperation;   // resume/abort offer data, per ref
  })[];
  faulted: boolean;
}

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

interface RunStartEvent {
  type: "run_start";            // was: agent_start
  runId: string;
}

interface RunResumeEvent {
  type: "run_resume";           // new
  runId: string;
}

interface RunCancelEvent {
  type: "run_cancel";           // was: abort
  runId: string;
  clearedSteer: AgentMessage[];
  clearedFollowUp: AgentMessage[];
}

interface RunEndEvent {
  type: "run_end";              // was: agent_end + settled
  runId: string;
  outcome: "completed" | "aborted" | "failed";
  /** Same definition as RunResult.finalMessage. Transcript deltas were
      already delivered via message/entry events; agent_end.messages is gone. */
  finalMessage?: AssistantMessage;
  error?: ErrorInfo;
}

interface FaultEvent {
  type: "fault";                // new
  code: string;
  message: string;
}

// was: coding-agent ExtensionError via onError. One channel for all
// extension-code failures — hook handlers and event listeners (section 10).
type HandlerErrorEvent = {
  type: "handler_error";
  runId?: string;
  error: string;
  stack?: string;
} & (
  | { kind: "hook"; hook: string }        // hook type
  | { kind: "event"; event: string }      // event type being delivered
);

// Step and retry ----------------------------------------------------------

interface StepStartEvent {
  type: "step_start";           // was: turn_start
  runId: string;
  stepId: string;
}

interface StepEndEvent {
  type: "step_end";             // was: turn_end (also the save-point moment)
  runId: string;
  stepId: string;
  message: AssistantMessage;
  toolResults: ToolResultMessage[];
}

// Retry — unifies harness retry_scheduled/retry_attempt_start/retry_finished
// and coding-agent auto_retry_* / summarization_retry_*. Normal requests emit
// nothing: the UI's busy state spans run_start..run_end (or the
// compaction/navigation brackets). Retry events only surface the exception.

interface RetryScheduledEvent {
  type: "retry_scheduled";      // a request failed, next attempt is pending
  runId: string;
  purpose: "step" | "compaction" | "branch_summary";
  attempt: number;
  maxAttempts: number;
  delayMs: number;
  errorMessage: string;
}

interface RetryStartEvent {
  type: "retry_start";          // the scheduled attempt begins
  runId: string;
  purpose: "step" | "compaction" | "branch_summary";
  attempt: number;
}

interface RetryEndEvent {
  type: "retry_end";            // retrying resolved: success, or final failure
  runId: string;
  purpose: "step" | "compaction" | "branch_summary";
  attempt: number;
  success: boolean;
  finalError?: string;
}

// Messages and tools ------------------------------------------------------

interface MessageStartEvent {
  type: "message_start";
  runId?: string;               // absent for idle SessionTree writes
  message: AgentMessage;
}

interface MessageUpdateEvent {
  type: "message_update";       // only emitted for assistant messages during streaming, as today.
  runId: string;                // AssistantMessageEvent already carries the partial message and
  message: AgentMessage;        // per-block deltas; we add nothing on top.
  assistantMessageEvent: AssistantMessageEvent;
}

interface MessageEndEvent {
  type: "message_end";
  runId?: string;
  message: AgentMessage;
  entryId: string;              // new: the committed session entry
}

interface ToolStartEvent {
  type: "tool_start";           // was: tool_execution_start
  runId: string;
  stepId: string;
  toolCallId: string;
  toolName: string;
  args: unknown;                // effective args, after before_tool
}

interface ToolUpdateEvent {
  type: "tool_update";          // was: tool_execution_update
  runId: string;
  stepId: string;
  toolCallId: string;
  toolName: string;
  partialResult: AgentToolResult<unknown>;
}

interface ToolEndEvent {
  type: "tool_end";             // was: tool_execution_end
  runId: string;
  stepId: string;
  toolCallId: string;
  toolName: string;
  result: AgentToolResult<unknown>;
  isError: boolean;
}

// Session and config ------------------------------------------------------

interface EntryAddedEvent {
  type: "entry_added";          // generalizes coding-agent entry_appended
  entry: SessionTreeEntry;      // non-message entries: custom, label, name, config, compaction, summary
}

interface WritePendingEvent {
  type: "write_pending";        // new — deferred write durably accepted
  runId: string;
  entryId: string;              // provisioned id; entry_added/message_end follows with the same id
  entry: SessionTreeEntry;
}

interface QueueUpdateEvent {
  type: "queue_update";         // per ref — like every event it carries ref;
  steer: AgentMessage[];        // these are that ref's queues
  followUp: AgentMessage[];
  nextRun: AgentMessage[];
}

// Payloads identify the change compactly; clients needing full objects use the
// getters (locally or via their server). streamOptions carries no value: headers
// may hold secrets, transport may be a function.
type ConfigUpdateEvent = {
  type: "config_update";        // was: model_update, thinking_level_update, tools_update, resources_update
} & (
  | { property: "model"; value: { provider: string; modelId: string }; previous: { provider: string; modelId: string } | null }
  | { property: "thinkingLevel"; value: ThinkingLevel; previous: ThinkingLevel }
  | { property: "activeTools"; value: string[]; previous: string[] }
  | { property: "tools"; value: string[]; previous: string[] }          // names
  | { property: "resources"; value: { skills: string[]; promptTemplates: string[] } }  // names
  | { property: "streamOptions" }
  | { property: "retryPolicy"; value: RetryPolicy }
  | { property: "compactionSettings"; value: CompactionSettings }
  | { property: "steeringMode"; value: QueueMode }
  | { property: "followUpMode"; value: QueueMode }
);

// Compaction and navigation ----------------------------------------------

interface CompactionStartEvent {
  type: "compaction_start";     // from coding-agent compaction_start
  runId: string;                // the run for auto, the operation for manual
  reason: "manual" | "threshold" | "overflow";
}

// End events mirror operation_finished.outcome and the result types — one
// vocabulary across log, results, and events.
interface CompactionEndEvent {
  type: "compaction_end";       // was: session_compact + coding-agent compaction_end
  runId: string;
  reason: "manual" | "threshold" | "overflow";
  outcome: "completed" | "cancelled" | "aborted" | "failed";
  entry?: CompactionEntry;      // outcome "completed"
  fromHook: boolean;
  error?: ErrorInfo;            // outcome "failed"
}

interface NavigationStartEvent {
  type: "navigation_start";     // new — operation accepted, summary may generate
  runId: string;                // the navigation operation
  targetId: string;
}

interface NavigationEndEvent {
  type: "navigation_end";       // was: session_tree — the leaf moves atomically here;
  runId: string;                // navigateTree() is the only cursor mutation
  outcome: "completed" | "cancelled" | "aborted" | "failed";
  oldLeafId: string | null;
  newLeafId: string | null;
  summaryEntry?: BranchSummaryEntry;   // outcome "completed", when summarize
  error?: ErrorInfo;                   // outcome "failed"
}

run_start
  step_start
    message_start
    message_update*
    message_end          assistant committed
    tool_start / tool_update* / tool_end     per tool call
    message_end          toolResult committed, source order
  step_end
  compaction_start       auto-compaction at a checkpoint, when needed
    entry_added          compaction entry committed
  compaction_end
  step_start ... step_end                    until no continuation
run_end

compaction_start         reason: manual        navigation_start
  entry_added            compaction entry        entry_added        summary entry
compaction_end                                  navigation_end       leaf moves here

const off = harness.hooks.on("before_tool", async (event) => {
  if (event.toolName === "bash") return { block: { reason: "not allowed" } };
});

// Run boundaries ----------------------------------------------------------

// was: before_agent_start. Once per run, before durable acceptance.
// Not re-run on retry or resume; its effective output is persisted.
//
// Durable run setup, unlike transform_context: returned messages become
// session entries after the prompt (skill preambles, injected context files);
// the effective system prompt is stored in the operation-start harness entry
// and used for the whole run, including resume. transform_context is
// per-request and ephemeral: it shapes what the provider sees, never what
// the session contains.
interface BeforeRunHook {
  event: {
    prompt: (TextContent | ImageContent)[];
    systemPrompt: string;
    resources: AgentHarnessResources;
  };
  result: {
    /** Persisted as session entries after the prompt. */
    messages?: AgentMessage[];
    /** Persisted as systemPromptOverride in the operation-start harness entry;
        fixed for the whole run. Without an override, the systemPrompt config
        callback is evaluated per request instead. */
    systemPrompt?: string;
    /** Opaque JSON, keyed by extension id, persisted in the operation-start
        harness entry, handed back to before_resume (possibly on another
        machine). For per-run state that would otherwise live in a closure:
        external job ids, idempotency keys, mode flags. Keep it small. */
    resumeData?: JsonValue;
  } | undefined;
}

// New. On resume(), before any effect. Rebuilds process-local extension
// state; must be idempotent (a crash can rerun it). Cannot rewrite the
// accepted prompt or system prompt.
interface BeforeResumeHook {
  event: {
    runId: string;
    kind: "run" | "compaction" | "navigation";
    /** Persisted effective before_run output. */
    prepared: { prompt: (TextContent | ImageContent)[]; systemPromptOverride?: string };
    resumeData?: JsonValue;
  };
  result: void;
}

// was: the actionable part of agent_end/settled. Runs when nothing is
// pending: no tool continuation, no queued messages. Work enqueued here
// (returned or via followUp()) continues the same run: no new
// run_start/run_end pair, same runId, more steps. run_end fires once,
// when this boundary passes with nothing pending.
interface BeforeRunEndHook {
  event: { runId: string; messages: AgentMessage[] };
  result: { followUp?: string } | undefined;   // or call steer()/followUp() directly
}

// Request pipeline ---------------------------------------------------------

// was: context. AgentMessage level, before convertToLlm.
// Pruning, injection, custom-message handling.
interface TransformContextHook {
  event: { messages: AgentMessage[] };
  result: { messages: AgentMessage[] } | undefined;
}

// was: before_provider_request. Provider-neutral request, after conversion.
interface BeforeRequestHook {
  event: {
    model: Model<any>;
    purpose: "step" | "compaction" | "branch_summary";
    attempt: number;
    streamOptions: AgentHarnessStreamOptions;
  };
  result: { streamOptions?: AgentHarnessStreamOptionsPatch } | undefined;
}

// was: before_provider_payload. Provider-specific wire payload. Last stop.
interface BeforePayloadHook {
  event: { model: Model<any>; payload: unknown };
  result: { payload: unknown } | undefined;
}

// was: after_provider_response (observation) + message_end replacement
// (mutation). Runs after the stream finishes, before the assistant message
// is committed. The committed message is what events and the session see.
interface AfterResponseHook {
  event: {
    status: number;
    headers: Record<string, string>;
    message: AssistantMessage;
  };
  result: { message?: AssistantMessage } | undefined;   // must keep role
}

// Tools --------------------------------------------------------------------

// was: tool_call + loop beforeToolCall. After validation, before execution.
// Effective args are persisted in the tool_started harness entry.
interface BeforeToolHook {
  event: {
    toolCallId: string;
    toolName: string;
    args: Record<string, unknown>;
  };
  result: {
    args?: Record<string, unknown>;
    block?: { reason: string };
  } | undefined;
}

// was: tool_result + loop afterToolCall. Patch semantics field-by-field,
// no deep merge, as today.
interface AfterToolHook {
  event: {
    toolCallId: string;
    toolName: string;
    args: Record<string, unknown>;
    content: (TextContent | ImageContent)[];
    details: unknown;
    isError: boolean;
    usage?: Usage;
  };
  result: {
    content?: (TextContent | ImageContent)[];
    details?: unknown;
    isError?: boolean;
    usage?: Usage;
    terminate?: boolean;
  } | undefined;
}

// Structural operations ----------------------------------------------------

// was: session_before_compact. Cancel, adjust, or supply the summary.
interface BeforeCompactionHook {
  event: {
    reason: "manual" | "threshold" | "overflow";
    preparation: CompactionPreparation;
    customInstructions?: string;
  };
  result: { cancel?: boolean; compaction?: CompactResult } | undefined;
}

// was: session_before_tree. Cancel, adjust, or supply the branch summary.
interface BeforeNavigationHook {
  event: { targetId: string; preparation: TreePreparation };
  result: {
    cancel?: boolean;
    summary?: { summary: string; details?: unknown; usage?: Usage };
    customInstructions?: string;
    replaceInstructions?: boolean;
    label?: string;
  } | undefined;
}

H   hook (awaited)
E   event (passive)
CS  durable append: session entry (tree)
CH  durable append: harness entry (orchestration, schemas: section 5)
X   process dies

   prompt("fix the bug")
H  before_run                              may inject messages, transform system prompt
CH operation_started → E run_start         op-1, prepared output persisted
CS message user → E message_start, message_end
E  step_start                              step-1
H  transform_context
H  before_request
H  before_payload
CH generation_started                before the billable request
E  message_start                           assistant streaming begins
E  message_update*
H  after_response                          may replace the assistant message
CS message assistant [tool call] → E message_end
H  before_tool                             may mutate args or block
CH tool_started                            effective args persisted
E  tool_start
E  tool_update*
H  after_tool                              may patch the result
E  tool_end
CS message toolResult → E message_start, message_end
E  step_end
   checkpoint: deferred writes, queues, compaction — nothing pending
E  step_start                              step-2
H  transform_context / before_request / before_payload
CH generation_started
E  message_start, message_update*
H  after_response
CS message assistant "done" → E message_end
E  step_end
H  before_run_end                          returns nothing
CH operation_finished → E run_end          outcome: completed

E  tool_start                              tool executing
   steer("focus on tests")                 caller resolves at CH
CH queue_enqueued → E queue_update         tree-neutral, mid-step is fine
E  tool_end
CS message toolResult → E message_start, message_end
E  step_end
   checkpoint consumes steering
CS message user "focus on tests" → E message_start, message_end, queue_update
E  step_start                              next request sees the steering message

E  step_end
H  before_run_end                          handler calls followUp("now write tests")
CH queue_enqueued → E queue_update         durable before the hook returns
   run continues — same runId, no new run_start
CS message user "now write tests" → E message_start, message_end
E  step_start ...
H  before_run_end                          runs again after those steps; returns nothing
CH operation_finished → E run_end          exactly one, outcome: completed

CH generation_started                attempt 1
   provider fails (overloaded)
E  retry_scheduled                          attempt 1, delayMs
E  retry_start                              attempt 2
H  transform_context / before_request / before_payload    re-run per attempt
CH generation_started                attempt 2 — durable count survives restarts
E  message_start, message_update*
E  retry_end                                success: true
CS message assistant → E message_end

E  tool_start                              tool executing
   abort()                                 caller resolves after CH + signal
CH operation_cancelled → E run_cancel      cleared steer/follow-up items returned
   tool signalled; reconciliation in background
CS message toolResult → E message_start, message_end     synthetic "interrupted", or real if it finished
CS message assistant → E message_start, message_end      stopReason: aborted — provider-valid closure
CH operation_finished → E run_end          outcome: aborted

E  step_end
   checkpoint: prospective context too big
E  compaction_start                        reason: threshold
H  before_compaction                       may cancel or supply the summary
CH generation_started                purpose compaction — skipped if hook supplied
CS compaction entry → E entry_added
E  compaction_end
E  step_start                              next request uses compacted context

CS message assistant [tool call] → E message_end
CH tool_started                            replay: never
X  machine dies mid-execution

   — new machine —
   AgentHarness.create(...)                suspended: { kind: "run", ... }
H  before_resume                           receives persisted resumeData; idempotent
   resume()
E  run_resume
CS message toolResult → E message_start, message_end    synthetic "interrupted", not re-run; recovery: true
E  step_end                                recovery: true
E  step_start                              run continues normally from here
H  transform_context / before_request / before_payload
CH generation_started
...

CH generation_started                attempt 1
X  dies mid-stream — partial tokens lost, never persisted

   — restore + resume —
E  run_resume
H  transform_context / before_request / before_payload
CH generation_started                attempt 2: same step, durable attempt count
E  message_start ...

   navigateTree(target, { summarize: true })
CH operation_started → E navigation_start  destination + provisioned summary/leaf ids
H  before_navigation                       may cancel (→ finished cancelled) or supply the summary
CH generation_started                purpose branch_summary — skipped if hook supplied
CS leaf entry → E entry_added              cursor moves — the atomic boundary
CS branch_summary entry → E entry_added
CH operation_finished → E navigation_end   old/new leaf, summary

CH generation_started                      request in flight; context tip is user U1
   handler calls session.appendMessage(M)  a custom message that projects into context
CH write_deferred → E write_pending        durable acceptance; M is not in the tree yet
CS message assistant A1 → E message_end    provider cached prefix [.., U1, A1]
E  step_end
   checkpoint applies deferred writes
CS message M → E message_start, message_end   tail append, after A1
E  step_start                              next context [.., U1, A1, M, ...] — prefix intact

   main: prompt("fix the bug")     slack:t1: prompt("summarize this thread")
CH operation_started   ref=main
CH operation_started   ref=slack:t1
CH generation_started  ref=main       attempt 1 for main's cycle
CH generation_started  ref=slack:t1   attempt 1 for t1's cycle — no interference
CS message assistant                  chained on main's branch (no ref field —
CS message assistant                  chained on t1's branch    membership by parentId)
CH operation_finished  ref=slack:t1 → E run_end (ref slack:t1)
CH operation_finished  ref=main     → E run_end (ref main)

flowchart TD
    O[open session, enumerate refs] --> A{ref has unmatched
operation_started?}
    A -->|no| I[ref Idle]
    I --> NR[collect ref's pending next-run items]
    A -->|yes| T[read ref's tail: seq > opStart]
    T --> V{tail valid?}
    V -->|no| X[corruption error]
    V -->|yes| M[resolve model/tool identities]
    M --> S[ref Suspended]

interface HarnessState {
  /** One slot per ref; the structure below is per ref. */
  refs: Map<string, RefState>;
}

interface RefState {
  leafId: string | null;

  operation: null | {
    id: string;
    kind: "run" | "compaction" | "navigation";
    sourceLeafId: string | null;
    intent: OperationStartedEntry["intent"];
    cancelled: boolean;                          // operation_cancelled present

    /** Provisioned initial messages whose ids have no entry, in order. Runs only. */
    missingInitialMessages: ProvisionedMessage[];

    /** Attempts already made for the current request cycle — there is never
        more than one live. Incremented on generation_started; reset to 0
        whenever a session entry commits (the transcript advanced, the next
        request asks a different question). Restore seeds it during the tail
        reduction: same two rules applied left to right. */
    requestAttempts: number;

    /** This run's newest assistant message with tool calls, if any, with per-call state. */
    toolBatch: null | {
      assistantEntryId: string;
      calls: {
        toolIndex: number;
        toolCallId: string;
        started: ToolStartedEntry | null;
        resultExists: boolean;
      }[];
    };

    /** Accepted, unconsumed, in acceptance order. Dead (returned via suspended.cancelled) if cancelled. */
    pendingSteer: ProvisionedMessage[];
    pendingFollowUp: ProvisionedMessage[];
    /** Accepted, unapplied, in acceptance order. Survive cancellation. */
    pendingWrites: ProvisionedEntry[];

    /** Structural targets: does an entry with the provisioned id exist? */
    targets: { result?: boolean; leaf?: boolean; summary?: boolean; label?: boolean };
  };

  /** This ref's queue_enqueued(nextRun) after its last run-kind operation start, targets absent. */
  pendingNextRun: ProvisionedMessage[];
}

async function resume(): Promise<ResumeResult> {   // per ref
  if (suspended.missing.tools.length || suspended.missing.models.length) {
    return { ok: false, outcome: "rejected", error: { code: "missing_identities", message: ... } };
  }
  events.emit({ type: "run_resume", runId: op.id, recovery: true });
  switch (op.kind) {
    case "run":        return { kind: "run",        ...await runProcedure() };
    case "compaction": return { kind: "compaction", ...await compactionProcedure() };
    case "navigation": return { kind: "navigation", ...await navigationProcedure() };
  }
}

/** Append a provisioned session entry unless an entry with its id already exists. */
async function appendIfMissing(target: ProvisionedEntry): Promise<void> {
  if (!(await session.getEntry(target.id))) {
    await appendSessionEntry(target);        // → message/entry events, recovery-flagged during resume
  }
}

async function runProcedure(): Promise<RunResult> {
  try {
    // Initial messages — unconditional, even when cancelled below:
    // accepted content is never dropped.
    for (const msg of op.intent.initialMessages) await appendIfMissing(msg);

    if (state.cancelled) return await cancellationPath();

    if (state.toolBatch?.calls.some((c) => !c.resultExists)) {
      await reconcileToolBatch(state.toolBatch);
    }

    return await driverLoop();
  } catch (err) {
    return await handleRunError(err);
  }
}

async function handleRunError(err: unknown): Promise<RunResult> {
  if (err instanceof RunFailed) {
    await appendHarnessEntry({ type: "operation_finished", outcome: "failed", error: err.info });
    // newestAssistantProjection() is run-scoped: newest message entry appended
    // after operation_started that projects to an AssistantMessage. May be undefined.
    const finalMessage = newestAssistantProjection();
    events.emit({ type: "run_end", runId: op.id, outcome: "failed", error: err.info, finalMessage });
    return { ok: false, outcome: "failed", runId: op.id, error: err.info, finalMessage };
  }
  enterFaultedState(err);   // AppendFailed, or a bug — either way we cannot safely continue
  events.emit({ type: "fault", code: ..., message: ... });
  return { ok: false, outcome: "faulted", runId: op.id, error: errorInfo(err) };
}

async function driverLoop(): Promise<RunResult> {
  while (true) {
    // ── checkpoint ─────────────────────────────────────────────
    for (const write of state.pendingWrites) await appendIfMissing(write.target);
    for (const msg of takeQueued(state.pendingSteer, config.steeringMode)) {
      await appendIfMissing(msg);
    }
    if (await contextOverLimit()) await autoCompact();      // may throw RunFailed

    // ── step, while the model owes a response ─────────────────────────
    if (needsAssistantResponse()) {          // newest run message is user/steering/toolResult
      const assistant = await requestAssistant();           // may throw RunFailed
      if (hasToolCalls(assistant)) await executeToolBatch(assistant);
      continue;                              // every step ends in a fresh checkpoint
    }

    // ── follow-ups ──────────────────────────────────────────────
    const followUps = takeQueued(state.pendingFollowUp, config.followUpMode);
    if (followUps.length > 0) {
      for (const msg of followUps) await appendIfMissing(msg);
      continue;
    }

    // ── finish boundary ────────────────────────────────────────
    const result = await hooks.run("before_run_end", { runId: op.id, messages: runMessages() });  // this run's messages
    if (result?.followUp) await harness.followUp(result.followUp);
    if (hasPendingWork()) continue;          // hook enqueued something → keep going

    await appendHarnessEntry({ type: "operation_finished", outcome: "completed" });
    const finalMessage = newestAssistantProjection();
    events.emit({ type: "run_end", runId: op.id, outcome: "completed", finalMessage });
    return { ok: true, runId: op.id, finalMessage };
  }
}

async function requestAssistant(): Promise<AssistantMessage> {
  while (true) {
    // Retrying the same position grows the count; any committed session entry
    // resets it. Seeded from the log at restore, so the bound survives
    // restarts (see HarnessState.requestAttempts).
    const attempt = state.requestAttempts + 1;
    if (attempt > config.retry.maxAttempts) {
      await appendSessionEntry(errorAssistantMessage());    // transcript records the give-up
      throw new RunFailed({ code: "retries_exhausted", message: ... });
    }

    // Effective system prompt, per request: the persisted before_run override
    // if set, else the systemPrompt config callback evaluated fresh — it sees
    // current active tools, preserving the old mid-run rebuild behavior.
    const systemPrompt = op.intent.systemPromptOverride ?? await evalSystemPromptConfig();
    const context = await hooks.run("transform_context", { messages: await contextMessages() });
    const options = await hooks.run("before_request", { model, purpose: "step", attempt, streamOptions });
    // before_payload runs inside the provider call, on the wire payload

    await appendHarnessEntry({ type: "generation_started", purpose: "step", attempt, ... });
    try {
      const response = await streamRequest(context, options);        // → message_start/update events
      const final = (await hooks.run("after_response", response))?.message ?? response.message;
      await appendSessionEntry(assistantEntry(final));               // → message_end
      return final;
    } catch (err) {
      if (!isRetryable(err)) {
        await appendSessionEntry(errorAssistantMessage(err));
        throw new RunFailed(errorInfo(err));
      }
      events.emit({ type: "retry_scheduled", attempt, delayMs: backoff(attempt), ... });
      await sleep(backoff(attempt));
      events.emit({ type: "retry_start", attempt: attempt + 1, ... });
    }
  }
}

async function autoCompact(): Promise<void> {
  events.emit({ type: "compaction_start", runId: op.id, reason: "threshold" });
  const prep = prepareCompaction(await contextEntries());
  const hook = await hooks.run("before_compaction", { reason: "threshold", preparation: prep });
  if (hook?.cancel) {
    events.emit({ type: "compaction_end", runId: op.id, outcome: "cancelled", ... });
    return;                                  // run continues; overflow, if it comes, fails the step
  }
  const result = hook?.compaction ?? await generateBounded("compaction", prep);  // may throw RunFailed
  await appendSessionEntry(compactionEntry(result));       // → entry_added
  events.emit({ type: "compaction_end", runId: op.id, outcome: "completed", entry, fromHook: !!hook?.compaction, ... });
}

async function reconcileToolBatch(batch: ToolBatch): Promise<void> {
  for (const call of batch.calls) {                        // assistant source order
    if (call.resultExists) continue;                       // committed, incl. hook-patched content

    if (call.started) {
      // tool_started exists ⇒ before_tool and validation already ran and
      // cleared this invocation; effective args and the not-blocked decision
      // are the durable outcome.
      if (call.started.replay === "safe") {
        const result = await executeTool(call.started.toolName, call.started.effectiveArgs);
        const patched = await hooks.run("after_tool", { ...call, ...result });
        await appendIfMissing(toolResultEntry(call.started.resultEntryId, patched ?? result));
      } else {
        // replay "never": the effect may or may not have happened; running it
        // again is worse than admitting that. No hooks run.
        await appendIfMissing(syntheticToolResult(call.started.resultEntryId, "interrupted"));
      }
    } else {
      // never began: full normal path
      await executeToolCallNormally(call);   // validate → before_tool → block? error result
                                             // : tool_started → execute → after_tool → result
    }
  }
  // The step then ends normally; the model sees every call answered.
}

async function cancellationPath(): Promise<RunResult> {
  // initial messages were already appended by runProcedure()

  for (const call of state.toolBatch?.calls ?? []) {       // source order
    if (call.resultExists) continue;
    await appendIfMissing(syntheticToolResult(
      call.started ? call.started.resultEntryId : provisionResultId(call),
      call.started ? "interrupted" : "aborted",            // post-crash there is nothing to salvage
    ));
  }

  for (const write of state.pendingWrites) await appendIfMissing(write.target);  // facts survive abort

  // The transcript always records how the run ended — including the
  // no-assistant-yet case (cancelled during the first request).
  if (!newestRunAssistantIsAborted()) {
    await appendSessionEntry(abortedClosureMessage());
  }

  await appendHarnessEntry({ type: "operation_finished", outcome: "aborted" });
  const finalMessage = newestAssistantProjection();        // the aborted closure
  events.emit({ type: "run_end", runId: op.id, outcome: "aborted", finalMessage });
  return { ok: false, outcome: "aborted", runId: op.id, finalMessage };
}

async function compactionProcedure(): Promise<CompactionRunResult> {
  try {
    // op.persisted is runtime-only (not in HarnessState): false when the live
    // call enters before its operation_started entry exists, true on resume.
    if (!op.persisted) await appendOperationStarted();               // live entry point
    events.emit({ type: "compaction_start", runId: op.id, reason }); // re-emitted on resume

    if (!state.targets.result) {
      const hook = await hooks.run("before_compaction", { reason, preparation, customInstructions });
      if (hook?.cancel) {
        await appendHarnessEntry({ type: "operation_finished", outcome: "cancelled" });
        events.emit({ type: "compaction_end", runId: op.id, outcome: "cancelled", ... });
        return { ok: false, outcome: "cancelled", runId: op.id };
      }
      const result = hook?.compaction
        ?? await generateBounded("compaction", preparation);         // may throw RunFailed
      await appendSessionEntry(compactionEntry(op.intent.resultEntryId, result));
    }

    await appendHarnessEntry({ type: "operation_finished", outcome: "completed" });
    events.emit({ type: "compaction_end", runId: op.id, outcome: "completed", entry, ... });
    return { ok: true, runId: op.id, entry };
  } catch (err) {
    return await handleStructuralError(err);
  }
}

// Structural twin of handleRunError. The failed path still emits the end
// event so the start/end bracket balances for every outcome.
async function handleStructuralError(err: unknown) {
  const endEvent = op.kind === "compaction" ? "compaction_end" : "navigation_end";
  if (err instanceof RunFailed) {
    await appendHarnessEntry({ type: "operation_finished", outcome: "failed", error: err.info });
    events.emit({ type: endEvent, runId: op.id, outcome: "failed", error: err.info, ... });
    return { ok: false, outcome: "failed", runId: op.id, error: err.info };
  }
  enterFaultedState(err);   // AppendFailed, or a bug
  events.emit({ type: "fault", code: ..., message: ... });
  return { ok: false, outcome: "faulted", runId: op.id, error: errorInfo(err) };
}

async function navigationProcedure(): Promise<NavigationRunResult> {
  try {
    if (!op.persisted) await appendOperationStarted();
    events.emit({ type: "navigation_start", runId: op.id, targetId });  // re-emitted on resume
    const { intent } = op;
    let summary: SummaryContent | undefined;

    if (!state.targets.leaf) {
      // the navigation has not happened yet
      const hook = await hooks.run("before_navigation", { targetId: intent.targetId, preparation });
      if (hook?.cancel) {
        await appendHarnessEntry({ type: "operation_finished", outcome: "cancelled" });
        events.emit({ type: "navigation_end", runId: op.id, outcome: "cancelled", ... });
        return { ok: false, outcome: "cancelled", runId: op.id };
      }
      summary = hook?.summary
        ?? (intent.summarize ? await generateBounded("branch_summary", ...) : undefined);
      await appendSessionEntry(leafEntry(intent.leafEntryId, intent.destinationLeafId));
      // ↑ the cursor moves here, atomically
    }

    if (intent.summarize && !state.targets.summary) {
      // leaf moved pre-crash without a summary → regenerate (attempt-bounded)
      summary ??= await generateBounded("branch_summary", ...);
      await appendIfMissing(summaryEntry(intent.summaryEntryId, summary));
    }
    if (intent.label && !state.targets.label) {
      await appendIfMissing(labelEntry(intent.labelEntryId, intent.label));
    }

    await appendHarnessEntry({ type: "operation_finished", outcome: "completed" });
    events.emit({ type: "navigation_end", runId: op.id, outcome: "completed", oldLeafId, newLeafId, summaryEntry, ... });
    return { ok: true, runId: op.id, newLeafId, summaryEntry };
  } catch (err) {
    return await handleStructuralError(err);
  }
}

interface SessionCreateOptions {
  id?: string;
  /** New. Link a fresh session to a parent, e.g. a subagent's session to the
      session whose tool call spawned it. Same linkage fork sets automatically. */
  parentSessionId?: string;
}

type SessionForkOptions =
  /** Existing behavior: the selected branch only — root to the fork point.
      Sibling branches are not copied. Default: the source main's branch. */
  | { scope?: "branch"; entryId?: string; position?: "before" | "at" }
  /** New: the entire tree — all session entries, every branch, leaf preserved. */
  | { scope: "tree" };

interface SessionRepo {
  ...
  create(options: TCreateOptions): Promise<Session>;
  fork(source, options: SessionForkOptions & TCreateOptions): Promise<Session>;
}

CREATE TABLE harness_entries (
  session_id TEXT NOT NULL,
  seq        INTEGER NOT NULL,      -- shared sequence with session_entries
  id         TEXT NOT NULL,
  type       TEXT NOT NULL,         -- operation_started, generation_started, ...
  ref        TEXT NOT NULL,         -- partition key for per-ref reduction
  run_id     TEXT NOT NULL,
  op_kind    TEXT,                  -- operation_started only: run | compaction | navigation
  timestamp  TEXT NOT NULL,
  payload    TEXT NOT NULL,
  PRIMARY KEY (session_id, seq)
);
CREATE UNIQUE INDEX idx_harness_session_id ON harness_entries(session_id, id);
CREATE INDEX idx_harness_ref_type_seq ON harness_entries(session_id, ref, type, seq);
CREATE INDEX idx_harness_ref_kind_seq ON harness_entries(session_id, ref, type, op_kind, seq);

interface PiObservabilityEvent {
  type: "start" | "end" | "error" | "event";
  name: string;                          // e.g. "pi.harness.generation"
  traceId: string;
  spanId?: string;
  parentSpanId?: string;
  timestamp: number;
  durationMs?: number;
  context?: Record<string, unknown>;     // user context, see below
  payload?: Record<string, unknown>;     // safe attributes, see redaction
  error?: { name: string; message: string };
}

// Runtime-agnostic core; adapters supply context propagation.
interface PiObservability {
  getContext(): PiObservabilityContext | undefined;
  runWithContext<T>(context: PiObservabilityContext, fn: () => T): T;
  emit(event: PiObservabilityEvent): void;
  hasSubscribers(): boolean;             // skip payload assembly when nobody listens
}

function configurePiObservability(o: PiObservability): void;
function subscribePiObservability(listener: (e: PiObservabilityEvent) => void): () => void;
function runWithPiContext<T>(userContext: Record<string, unknown>, fn: () => T): T;
function traceOperation<T>(name: string, payload: Record<string, unknown>, fn: () => T): T;

await Promise.all([
  runWithPiContext({ userId: "alice" }, () => harnessA.prompt("A")),
  runWithPiContext({ userId: "bob"   }, () => harnessB.prompt("B")),
]);

pi.harness.run            runId, sessionId, recovery
├─ pi.harness.step         stepId
│  ├─ pi.harness.generation   purpose, attempt, provider, model
│  │  └─ pi.ai.provider.request  physical request — emitted by packages/ai
│  │                            (several per generation for split-turn compaction)
│  └─ pi.harness.tool         toolName, toolCallId, replay
├─ pi.harness.checkpoint    deferred writes / queue consumption / compaction decision
└─ pi.harness.hook          hook type — the awaited control points

pi.harness.compaction     manual operation (auto-compaction nests under its run)
pi.harness.navigation
pi.harness.resume         wraps recovery work; child spans as above
pi.session.append         entry type, seq — storage-level timing

// prompt() and resume() both land here
async function executeRun(op: RunOperation): Promise<RunResult> {
  return traceOperation("pi.harness.run",
    { runId: op.id, sessionId, recovery: op.persisted },
    () => runProcedure());
}

// driverLoop(): one span per step, per generation, per tool
await traceOperation("pi.harness.step", { runId: op.id, stepId }, async () => {
  const assistant = await traceOperation("pi.harness.generation",
    { purpose: "step", attempt, provider: model.provider, model: model.id },
    () => requestAssistant());              // pi.ai.provider.request nests underneath
  if (hasToolCalls(assistant)) await executeToolBatch(assistant);
});

// executeToolBatch(), per call
await traceOperation("pi.harness.tool",
  { runId: op.id, toolName: call.name, toolCallId: call.id, replay },
  () => executeTool(call));

// hook dispatch and Session appends: same one-line wrapping

// packages/ai/src/models.ts
streamSimple(model, context, options): AssistantMessageEventStream {
  const stream = this.doStreamSimple(model, context, options);   // existing body
  if (!hasSubscribers()) return stream;                          // zero cost when idle

  // stream.result() is the existing final-message promise; the span becomes
  // one more awaiter. Provider errors are in-band in pi-ai (stopReason
  // "error"/"aborted" messages, not rejections), so span status derives from
  // the final message; nothing propagates to the caller.
  void traceOperation("pi.ai.provider.request", {
    api: model.api, provider: model.provider, model: model.id,
    sessionId: options?.sessionId, reasoning: options?.reasoning,
  }, async () => {
    const message = await stream.result();
    if (message.stopReason === "error" || message.stopReason === "aborted") {
      throw new ProviderSpanError(message);   // → "error" span event
    }
    return { stopReason: message.stopReason, ...safeUsage(message.usage) };
  }).catch(() => {});                         // span error recorded; never rethrown

  return stream;
}

// every span in this run carries the user context
await runWithPiContext({ userId, orgId, sessionId }, () => harness.prompt(text));

// console JSON sink
subscribePiObservability((e) => log.write(JSON.stringify(e)));

// OTel bridge, in its entirety
subscribePiObservability((e) => {
  if (e.type === "start") {
    spans.set(e.spanId!, tracer.startSpan(e.name, { attributes: flatten(e.payload) }));
  } else if (e.type === "end" || e.type === "error") {
    const span = spans.get(e.spanId!);
    if (!span) return;
    if (e.type === "error") span.setStatus({ code: SpanStatusCode.ERROR, message: e.error?.message });
    span.end();
    spans.delete(e.spanId!);
  }
});

// Interactive pi: single ref, nothing changes. AgentHarness implements
// AgentRef for main; suspended has 0 or 1 entries, always "main".
const { harness, suspended } = await AgentHarness.create({ session, models, model });
for (const s of suspended) await harness.ref(s.ref)!.resume();   // or offer resume/abort in UI
await harness.prompt("fix the bug");
await harness.steer("focus on the tests");
harness.setModel(opus);                     // main's branch-anchored config

// Slack bot: channel = session + main, each thread = a ref keyed by thread_ts.
const key = `slack:${threadTs}`;
const t = harness.ref(key) ?? (await harness.createRef(key, pingedEntryId)).ref;
await t.prompt("summarize this thread");    // parallel with main and other threads
await t.steer("shorter");
t.setModel(haiku);                          // this thread only
await t.navigateTree(earlierId);            // moves this thread's leaf only
await t.session.appendMessage(msg);         // appends to this thread's branch
await t.nextRun("also check the links");    // consumed by this thread's next run

// Thread renderer: scoped snapshot + only this thread's events.
const { snapshot, start, unsubscribe } = await t.watch();
render(snapshot.transcript);
start((event) => update(event));

// Dashboard / server: inventory + firehose, no transcripts.
const sess = await harness.watchSession();
for (const r of sess.snapshot.refs) {
  if (r.run?.status === "suspended") await harness.ref(r.name)!.resume();
}