// agent-runtime.ts — the durable turn engine and its seams.
//
// A pure contract layer (zero node:*): the engine depends ONLY on three seams —
// SessionStore, EventSink, Model. No SQLite, no HTTP, no platform. The SAME
// code runs over the native seam (@junejs/server's agent-native, on the host
// SQLite driver) and over a Cloudflare Durable Object (agent-durable). It is the sibling
// of agent.ts (the defineAction registry the runtime consumes), not a
// replacement.
//
// Durability model (log-replay + step-checkpoint):
//   • the `messages` log IS the session state — a fresh process rebuilds the loop
//     position purely from the log, so resume is automatic.
//   • modelStep/toolStep memoize into a `steps` table; a completed step is skipped
//     on replay. The checkpoint key ALWAYS carries the session dimension (the
//     SessionStore is session-scoped) so keys cannot leak across sessions.

import { fileRefsIn, type FileRef, type FileStore } from "./agent-files";
import type { Principal } from "./context";

// ── domain ──────────────────────────────────────────────────────────────────

// The server↔core contract counter (#94). @junejs/server asserts this at surface
// construction (AgentDurableObject / NativeRuntime) against the number IT was built
// for — so when a package manager nests a second, older core copy under server
// (regular-dep version skew), the mismatch fails at POWER-ON with both versions
// named, instead of mid-turn with something like "sink.emit is not a function".
// Bump ONLY when the server↔core runtime contract changes shape (a seam signature,
// an event/store type the server constructs against) — not on every release.
// v2: AgentSession.idle() (#174) — NativeRuntime calls it to decide eviction.
// v3: grantAnswer / AgentSession.pending() (#261) — the hosts decide { policy } answerers
//     with them; AgentSession.onDequeue / drain() (#263) — the Durable Object renders
//     held turns with them.
// v4: AgentSession.onAnnounce / flushAnnouncements() (#260) — the hosts deliver input
//     announcements with them.
// v5: SessionStore.reset(inTx) (#260) — the engine carries undelivered announcements across a
//     reset inside the reset's own transaction; a host store that ignored the callback would
//     drop them silently.
// v6: AgentSession.note() and the `note` Msg role (#262) — the Durable Object's /note route
//     and note() call it.
// v7: AgentSession.files, ToolContext.files and the file.attached event — the hosts install
//     the session's FileStore (docs/rfc-agent-files.md).
export const RUNTIME_API_VERSION = 7;

// `providerState` (#92) is OPAQUE round-trip state a model adapter may attach to a
// tool call: some providers require it replayed verbatim (Gemini 3+ returns a
// thoughtSignature per functionCall and rejects replays that omit it). Contract:
// written by the adapter when it builds the reply, stored on the assistant Msg with
// the rest of the call, and handed back UNTOUCHED when the transcript replays to the
// adapter — the engine never reads it, and it is never part of identity (step keys
// and dispatch use `id` alone). Without this field, adapters smuggle the state
// inside `id` — leaking it into every ledger keyed by callId and breaking on any
// id normalization.
export type ToolCall = { id: string; name: string; input: unknown; providerState?: string };
export type Msg =
  | { role: "user"; turnId: string; text: string }
  // A proactive turn's opening message (§9): a schedule / another channel / the agent itself
  // seeded this turn (no inbound user). A DISTINCT role — not a user Msg — so the durable
  // transcript honestly attributes who initiated it (`by`); the model adapter maps it to a
  // user/system message (providers needn't support a new role). See RFC decision #6.
  | { role: "trigger"; turnId: string; text: string; by: string }
  | { role: "assistant"; turnId: string; text: string; toolCalls: ToolCall[] }
  // `isError`: the tool threw — `result` is `{ error: message }`, and the model adapter marks
  // the block as an error (Anthropic `is_error`) so the model reads a failed call, not data.
  | { role: "tool"; turnId: string; toolCallId: string; name: string; result: unknown; isError?: true }
  // A note (#262): something that belongs in the agent's history without being a turn or the
  // agent's own words — an operator's reply during a take-over, traffic a channel observed.
  // Attributed (`by`, `kind`, `at`); the model adapter renders it as labelled context, never
  // as the agent's own message or the correspondent's. `turnId` is the note's own id (`n_…`):
  // every Msg keeps one, and no turn's filters match a note. Appended by AgentSession.note().
  | { role: "note"; turnId: string; by: string; kind: NoteKind; text: string; at: string };
// What a note is (#262). The two the email take-over and observe mode need, and any other
// string an app defines.
export type NoteKind = "operator_reply" | "observed" | (string & {});
export type ModelReply = { text: string; toolCalls: ToolCall[] };
export type ToolSpec = { name: string; description: string; input: unknown };

// A normalized inbound event — the platform-agnostic envelope a turn was triggered
// by. Defined at this (lowest) layer because ToolContext carries it: a channel's
// capability tool (e.g. slack_read_thread) defaults its target — channel / thread /
// message ts — from the CURRENT turn's event, so the model can call it with no args.
// A channel adapter maps its native payload (Slack Events API, Crisp hook, …) into
// this one shape; agent-config and channels re-export it for adapter authors.
//
// `kind` distinguishes a user message from a reaction (emoji) or an edit, so a
// channel subscribing to reaction_added/removed can route those as turns too — `text`
// is present for message/app_mention, `reaction` for the emoji events. `raw` is the
// untouched platform payload: an escape hatch for anything not yet normalized.
export type InboundEvent = {
  source: string;                               // the channel that produced it ("slack" / "crisp")
  kind: "message" | "app_mention" | "reaction_added" | "reaction_removed" | "message_changed" | "rating" | "state_changed";
  channelId: string;                            // the conversation container: slack channel id / crisp website id
  threadId?: string;                            // thread within it: slack thread_ts / crisp conversation session id
  teamId?: string;                              // workspace/tenant the event belongs to (slack team_id) —
                                                // chat.startStream needs it as recipient_team_id in channels
  ts: string;                                   // this event's message ts
  user?: { id: string; name?: string; attested?: boolean };
                                                // WHO the platform CLAIMS is speaking. `attested`: the
                                                // platform vouches for `id` cryptographically (a signed
                                                // Slack event) — unlike a claim anyone can forge (an
                                                // email From:). Only an attested user is a default answerer
                                                // of the turn's requestInput (see Answerers).
  // WHO the app VERIFIED is speaking — resolved by a channel identity seam (e.g.
  // crispChannel's resolveIdentity) from platform-verified evidence, never from the
  // payload alone. The same Principal the UI/MCP paths carry (ActionContext.user), so
  // one authorization model spans every dispatch path. Must be JSON-serializable: it
  // rides the event across the /turn RPC and survives a suspend checkpoint (`raw` is
  // stripped there; principal is kept). Absent = anonymous turn (see Tool.requiresPrincipal).
  principal?: Principal;
  text?: string;                                // message / app_mention carry text; reactions don't
  reaction?: { name: string; itemTs: string };  // WHICH emoji, on WHICH message
  rating?: { stars: number; comment?: string }; // rating events: the score the visitor left (CSAT)
  state?: string;                               // state_changed events: the new conversation state (e.g. "resolved")
  raw?: unknown;                                // untouched platform payload (escape hatch); may
                                                // be dropped crossing the /turn RPC if unserializable
};

// The Model seam — provider-agnostic and STREAMING-FIRST. A model yields a stream of
// deltas as it produces them; a non-streaming provider is the degenerate case (a stream
// of length 1: just `done`). `reasoning`/`text` are live tokens the engine forwards as
// TurnEvents; `done` carries the authoritative assembled ModelReply the engine checkpoints
// (the adapter owns assembly, so the engine never re-derives tool calls from partial text).
// `opts.system` is the per-turn system prompt, injected by the runtime (see withSystem).
// WHY generation stopped, normalized across providers. Every provider ships its own
// vocabulary for this — Anthropic's Messages API `stop_reason` (end_turn / max_tokens /
// stop_sequence / tool_use / pause_turn / refusal / model_context_window_exceeded, per
// @anthropic-ai/sdk ≥0.60's Message type) and Gemini's `candidates[].finishReason`
// (STOP / MAX_TOKENS / SAFETY / RECITATION / MALFORMED_FUNCTION_CALL / …, per
// ai.google.dev/api/generate-content#FinishReason) — and both document that an abnormal
// stop may come with EMPTY content. An adapter that ignores the field converts a
// truncated/filtered response into a graceful empty reply, which the engine then
// "completes" silently (the production failure mode this type exists to kill). `raw`
// preserves the provider's own value for diagnostics.
export type ModelFinish = {
  reason: "stop" | "max_tokens" | "content_filter" | "malformed_tool_call" | "refusal" | "other";
  raw?: string;
};

export type ModelDelta =
  | { type: "reasoning"; text: string }   // thinking tokens (when the provider streams them)
  | { type: "text"; text: string }        // answer tokens
  | { type: "done"; reply: ModelReply; finish?: ModelFinish };  // terminal: the full assembled reply + why it stopped
export type Model = (msgs: Msg[], tools: ToolSpec[], opts?: { system?: string; systemMode?: "append" | "replace" }) => AsyncIterable<ModelDelta>;

// Per-source turn policy (#149): what a channel contributes to turns arriving
// through it, beyond transport. `overlay` is appended to the base system prompt
// (or REPLACES it under mode "replace" — for surfaces whose behavior genuinely
// isn't the base agent's, e.g. an operator-assist mode over a shadow judge).
// `denyTools` removes tools from the turn MECHANICALLY: not listed to the model,
// not dispatchable — policy the base prompt would otherwise have to beg for in
// prose ("do NOT call X"), now unpersuadable and prompt-injection-proof.
export type ChannelPolicy = {
  overlay?: string;
  overlayMode?: "append" | "replace";
  denyTools?: string[];
};

// A one-shot model output: a single-element stream. The degenerate streaming case, for
// scripted/test models and providers without token streaming.
export function replyStream(reply: ModelReply, finish?: ModelFinish): AsyncIterable<ModelDelta> {
  // Spread, don't assign: a no-claim call must not add an own `finish: undefined` property —
  // the pre-finish delta shape stays byte-identical for presence checks and deep equality.
  return (async function* () { yield { type: "done", reply, ...(finish ? { finish } : {}) }; })();
}

// Wrap a Model so every call carries `system` (the agent's instructions). The
// runtime applies this from the agent def, so a provider model needn't bake the
// system prompt in at construction — one model instance can serve many agents,
// each supplying its own system per turn. A per-turn `opts.system` (e.g. a channel
// overlay derived from the inbound event's source) is APPENDED to the base by
// default — the base instructions and the per-turn overlay compose. Under
// `systemMode: "replace"` the overlay IS the turn's system prompt: the base is
// dropped mechanically instead of the overlay pleading "disregard the above".
// systemMode is consumed here — the wrapped model only ever sees a final system.
export function withSystem(model: Model, system: string): Model {
  return (msgs, tools, opts) => {
    const { systemMode, ...rest } = opts ?? {};
    const finalSystem =
      systemMode === "replace" && opts?.system !== undefined
        ? opts.system // replace consumes ANY supplied overlay, empty string included
        : [system, opts?.system].filter(Boolean).join("\n\n");
    return model(msgs, tools, { ...rest, system: finalSystem });
  };
}

// A tool's `run` gets a context: its session-local `store` (write app state in
// the SAME tx as the checkpoint → exactly-once), and the `runtime` so a tool can
// spawn a child session — that is how a SUBAGENT works (a tool that runs a child
// actor). `run` is sync for local tools, async for remote ones (incl. subagents).
export interface ToolContext {
  store: SessionStore;
  runtime: Runtime;
  agent: string;
  sessionId: string;
  callId: string;
  // The inbound event that triggered this turn (when the turn came from a channel
  // that supplies one). A channel capability tool reads it to default its target —
  // e.g. slack_read_thread with no args reads ctx.event.threadId. Undefined for turns
  // not driven by a channel envelope (a bare /message POST, a scripted test).
  event?: InboundEvent;
  // The turn's resolved TRUSTED identity — a convenience mirror of event.principal
  // (see InboundEvent). A tool that reads tenant/user-scoped data keys its queries and
  // credentials off THIS, never off model-supplied input or event.user: the model
  // cannot influence it, so a prompt injection cannot steer a tool across tenants.
  principal?: Principal;
  // WHO OPENED this session (#128): the first resolved principal any turn arrived with,
  // recorded durably and immutable thereafter — in a multi-participant thread this lets
  // a tool tell the session's opener (initiator) apart from the current turn's speaker
  // (principal), e.g. only the initiator may widen a query's scope.
  // Deliberately NOT part of the requiresPrincipal gate: tool visibility keys off the
  // CURRENT speaker, so an anonymous follow-up in an operator-opened session does not
  // inherit the operator's tools. Undefined until some turn resolves a principal.
  initiator?: Principal;
  // Ask for external (human) input and SUSPEND the turn until session.resume provides it.
  // First call throws SuspendSignal to park the turn (durably); on the replay after resume it
  // returns the stored answer. Only usable from a remote tool (see Tool.mode) — a local tool
  // commits in the same tx and cannot park, so calling it there throws a FatalToolError (the
  // turn fails loudly). `answerers` says who may answer (see Answerers); it defaults to the turn's
  // speaker only when the channel attests that identity.
  requestInput(req: { id: string; prompt: string; schema?: unknown; answerers?: Answerers }): Promise<unknown>;
  // Where a tool puts a file it produces (docs/rfc-agent-files.md): put() stores the bytes
  // and returns a FileRef; return the ref in the result and the engine emits file.attached
  // for the channel to deliver. Always present: on a session whose host installed no store,
  // put() fails the turn with a message naming the missing configuration.
  files: FileStore;
}
export type Tool = {
  spec: ToolSpec;
  run: (input: any, ctx: ToolContext) => unknown;
  subagent?: boolean;
  // Principal gate: when true, the tool exists only on turns with a resolved trusted
  // identity (event.principal) — on an anonymous turn it is dropped from the model's
  // tool list entirely, so the model can neither call it nor see it. Mark every tool
  // that reads user/tenant-scoped data; leave knowledge/utility tools unmarked.
  requiresPrincipal?: boolean;
  // How the engine runs it (#233). "local": inside the store tx, exactly-once, and `run` must
  // return its result synchronously. "remote": awaited outside the tx, at-least-once, and the
  // only mode that can `ctx.requestInput`. Absent = derived from `run`: an `async` function is
  // remote, anything else local. Set it when `run` is async but isn't declared `async` — a
  // wrapper (`withRetry(async …)`), a plain function returning a Promise, transpiled code.
  mode?: ToolMode;
};
export type ToolMode = "local" | "remote";

// The mode the engine runs `run` in: the declared one, else remote exactly when `run` is an
// `async` function. Shared with actionToTool so a bridged action classifies the same way.
export function toolMode(run: Function, mode?: ToolMode): ToolMode {
  return mode ?? (run.constructor.name === "AsyncFunction" ? "remote" : "local");
}

function isThenable(x: unknown): x is PromiseLike<unknown> {
  return (typeof x === "object" || typeof x === "function") && x !== null && typeof (x as { then?: unknown }).then === "function";
}

// ── the two inner seams (LOCAL, per-session, co-located with execution) ───────
//
// A SessionStore instance is ALREADY scoped to one session — there is no
// session_id parameter anywhere. That is the structural fix for cross-session
// step-id collision: sessions cannot share checkpoint keys because they don't
// share a store. `tx` is a SYNCHRONOUS transaction: the durability contract
// (side effect + checkpoint + message append committing atomically) requires it.
export interface SessionStore {
  appendMessage(m: Msg): void;
  messages(): Msg[];
  // Has this turn's OPENING message already been appended? (a user msg for an inbound turn,
  // a trigger msg for a proactive one.) Guards against double-seeding on a crash-replay.
  hasOpeningMessage(turnId: string): boolean;
  getStep(id: string): unknown | undefined;
  putStep(id: string, output: unknown): void;
  delStep(id: string): void; // remove a checkpoint (a consumed suspend park); no-op when absent
  getStatus(): string;
  setStatus(s: string): void;
  tx<T>(fn: () => T): T; // synchronous transaction
  // Retire the session's current history (#129): ARCHIVE messages + steps under the
  // current generation number (never delete — the archive is the audit trail), clear the
  // live tables, set status back to "new", and return the archived generation index
  // (0-based). Must commit atomically (one tx). Optional: a store that doesn't implement
  // it makes AgentSession.reset() fail loudly instead of half-clearing.
  // `inTx`, when given, runs inside that same transaction AFTER the archive, so its writes
  // land in the fresh generation and commit (or roll back) with the reset — the engine
  // carries undelivered input announcements across it this way (#260).
  reset?(inTx?: () => void): number;
  // Escape hatch to the underlying storage handle, so a local tool can write its
  // own app table inside `tx` (exactly-once). On native this is the host sync
  // SQLite handle; on an edge target it is ctx.storage.sql.
  unwrap<H = unknown>(): H;
}

// ── the turn as a live event stream (see docs/rfc-turn-as-live-process.md) ────
// A turn emits a stream of typed events as it runs, observed via AgentSession.observe.
// Events split into STRUCTURAL (foldable from the durable log: message.completed,
// action.requested/completed, turn.completed) and LIVE-ONLY (reasoning.delta /
// message.delta, turn.started with its `trigger`, turn.failed, turn.cancelled, and
// input.requested). On a crash-replay a cached step short-circuits WITHOUT re-emitting,
// so a fresh subscriber sees only what runs from then on — UNLESS it attaches with
// observe({ turnId, replay: true }), which folds the structural prefix from the log
// before subscribing live (catch-up for a late or reconnecting subscriber). That
// catch-up is IN-PROCESS only: runStream, the Durable Object's SSE, and delivered
// renders subscribe live at turn start, and no host exposes a reconnectable events
// endpoint yet. A pending input.requested is not folded either — a late subscriber to a
// parked turn learns of the park from result()/the session status, not the replay.
// WHO may answer a parked input (#261).
//   { user }            — exactly this identity: compared with the resumer's VERIFIED `by`, a
//                         platform id such as the user id of a signature-checked Slack click.
//   { policy, scope? }  — a rule only the app can decide ("the operators of mailbox scout",
//                         "a manager of this tenant"): the host evaluates it at resume time
//                         through the agent's authorizeAnswer hook (grantAnswer) and hands the
//                         engine the grant; the engine checks the grant names exactly this rule.
// When a tool names none, the default is the turn's speaker if the channel ATTESTS that
// identity (event.user.attested); an inbound turn without one refuses to park (see
// requestInput) rather than park an input nobody can answer. A turn with no inbound event
// (proactive, programmatic) keeps no restriction — the app itself drives its resume.
export type Answerers = { user: string } | { policy: string; scope?: unknown };

// A tool's request for external (human) input that suspends the turn. `id` keys the answer
// (stable within the turn); `answerers` is who may answer (above). Serializable (it rides
// input.requested over SSE and persists in the suspended checkpoint).
export type InputRequest = { id: string; prompt: string; schema?: unknown; answerers?: Answerers };

// The app's decision for a { policy } answerer, made at resume time. `by` is the resumer's
// verified platform id, `principal` its app identity when the resuming surface resolved one
// (a bearer-authenticated CLI call, a signed-in page). Async: a rule may need the db.
export type AuthorizeAnswer = (a: {
  policy: string;
  scope?: unknown;
  by?: string;
  principal?: Principal;
  agent: string;
  session: string;
  request: InputRequest;
}) => boolean | Promise<boolean>;

// Canonical JSON (sorted keys) so a grant and a stored request compare by value, whatever
// the key order a host or a checkpoint round-trip produced.
function canonical(v: unknown): string {
  if (Array.isArray(v)) return `[${v.map(canonical).join(",")}]`;
  if (v && typeof v === "object") {
    return `{${Object.keys(v).sort().map((k) => `${JSON.stringify(k)}:${canonical((v as Record<string, unknown>)[k])}`).join(",")}}`;
  }
  return JSON.stringify(v) ?? "undefined";
}
function sameAnswerers(a: Answerers, b: Answerers): boolean {
  return canonical(a) === canonical(b);
}

// The answerers a parked input gets when its tool named none (see Answerers).
function defaultAnswerers(event: InboundEvent | undefined, inputId: string): Answerers | undefined {
  if (!event) return undefined;
  if (event.user?.attested) return { user: event.user.id };
  // The app must name who may answer — the model can't fix this, so it fails the turn.
  throw new FatalToolError(
    `requestInput("${inputId}"): this turn's speaker is not attested by its channel ("${event.source}"), so nobody could answer — ` +
      `name who may: ctx.requestInput({ …, answerers: { user } | { policy, scope } })`,
  );
}

// Thrown by ctx.requestInput to PARK a turn awaiting input. The engine catches it, persists
// the pending request as a checkpoint, and reports { status: "suspended" }. A later
// session.resume stores the answer and replays — the same durable step-checkpoint machine that
// gives exactly-once, now: "a step whose result comes from a human instead of a tool".
export class SuspendSignal extends Error {
  readonly request: InputRequest;
  readonly callId: string; // the tool call that parked the turn
  constructor(request: InputRequest, callId: string) {
    super(`turn suspended awaiting input: ${request.id}`);
    this.request = request;
    this.callId = callId;
  }
}

// A tool that throws does NOT fail the turn: the error becomes the tool's result, so the model
// can react — retry, try another way, tell the user (#232). Throw a FatalToolError for the rare
// error that must end the turn instead (a broken invariant, a misconfigured deployment): it
// propagates like before and the turn fails with it.
export class FatalToolError extends Error {
  constructor(message: string, options?: { cause?: unknown }) {
    super(message, options);
    this.name = "FatalToolError";
  }
}

// What the model reads for a thrown error: the message (with its class when it isn't a plain
// Error), never the stack, bounded so a huge message can't flood the context.
const TOOL_ERROR_MAX = 2000;
function toolErrorText(err: unknown): string {
  let text: string;
  try {
    text =
      err instanceof Error ? (err.name && err.name !== "Error" ? `${err.name}: ${err.message}` : String(err.message)) : typeof err === "string" ? err : String(err);
  } catch {
    // Anything can be thrown — `Object.create(null)` has no toString, a getter can throw.
    // Formatting must never be what fails the turn.
    text = "the tool threw a value that can't be converted to text";
  }
  return text.length > TOOL_ERROR_MAX ? `${text.slice(0, TOOL_ERROR_MAX)}…` : text;
}

// Thrown by AgentSession.resume when the resumer may not answer the pending request.
// A distinct class so a transport (the DO's /resume) can map it to 403 vs 409.
export class ResumeAuthorizationError extends Error {}

// Why a turn was cancelled (#129) — carried on CancelSignal and the turn.cancelled event
// so a consumer can pick honest copy: "replaced" (a newer message superseded it, the
// debounce path), "reset" (the session is being retired), "requested" (an explicit
// session.cancel with no replacement implied).
export type CancelReason = "replaced" | "reset" | "requested";

// Thrown by the engine when a cancellation request (session.cancel / a replace turn)
// takes effect — always at a CHECKPOINT BOUNDARY, never mid-step: everything committed
// stays committed, nothing dangles (a partially-run tool batch is closed with synthetic
// results first, so the transcript stays valid for the next model call). A distinct
// class, like SuspendSignal, so AgentSession.result maps it to { status: "cancelled" }
// instead of a failure. Cancellation is best-effort liveness, NOT a durability contract:
// the request lives in memory, so a crash-replay re-runs the turn uncancelled.
export class CancelSignal extends Error {
  readonly turnId: string;
  readonly reason: CancelReason;
  constructor(turnId: string, reason: CancelReason) {
    super(`turn ${turnId} was cancelled (${reason})`);
    this.turnId = turnId;
    this.reason = reason;
  }
}

// What the engine persists under the `suspended` step when a turn parks: enough to
// validate a later resume (turnId, the pending request) and replay the turn (userText/
// overlay/event). One per session — turns serialize, and start() rejects new turns while
// one is parked, so a single fixed key cannot be clobbered.
type SuspendedCheckpoint = { turnId: string; callId: string; request: InputRequest; userText: string; systemOverlay?: string; systemMode?: "append" | "replace"; deniedTools?: string[]; event?: InboundEvent };

// An inbound turn held while the session is parked (#263): everything start() needs to run it
// later, persisted in the session's own store (step INBOUND_QUEUE), so it survives a restart.
// `hostContext` is opaque to the engine — a host stores what it needs to finish the turn's
// delivery when it runs (e.g. "render through the source channel") and gets it back in
// onDequeue. `event.raw` is stripped, as in the suspended checkpoint.
export type QueuedTurn = { turnId: string; userText: string; event?: InboundEvent; trigger?: ProactiveTrigger; hostContext?: unknown; queuedAt: string };
const INBOUND_QUEUE = "inbound-queue";

// What a host hears about a session's parked input (#260), so it can keep a cross-session
// index of what waits on a person and update every surface showing a request:
//   parked   — a turn parked on requestInput (the request, and the event that triggered it);
//   held     — an inbound turn was held behind the park (#263): `queued` is the new count, the
//              approver's cue that the conversation moved on;
//   resolved — the park ended: "answered" by `by`, or "retired" by a session reset.
// Recorded in the SAME transaction as the state change it reports, in an outbox inside the
// session's store, and delivered after the commit; removed only once the host's hook
// returns. So delivery is at-least-once — `id` is unique per announcement; dedupe on it —
// and survives a crash between the commit and the hook (flushAnnouncements after a rebuild).
// The session stays the source of truth: an index built from these is a projection, and
// AgentSession.pending() is what to reconcile it against.
type AnnouncementBody =
  | { kind: "parked"; turnId: string; request: InputRequest; event?: InboundEvent; queued: number }
  | { kind: "held"; turnId: string; inputId: string; queued: number }
  | { kind: "resolved"; turnId: string; inputId: string; outcome: "answered" | "retired"; by?: string };
export type InputAnnouncement = { id: string; agent: string; session: string; at: string } & AnnouncementBody;
const ANNOUNCE_OUTBOX = "announce-outbox";

// How many held turns wait behind the park of `parkedTurnId` (#263): the parked turn itself is
// excluded — a held turn that ran and parked again stays in the held list until its turn
// settles (crash safety), and must not count as waiting behind itself. The ONE count used by
// pending().queued and by the parked / held announcements, so an index agrees with pending().
function heldBehind(store: SessionStore, parkedTurnId: string): number {
  return ((store.getStep(INBOUND_QUEUE) as QueuedTurn[] | undefined) ?? []).filter((q) => q.turnId !== parkedTurnId).length;
}

// Append to the outbox inside the CALLER's transaction (putStep is insert-only: replace).
function recordAnnouncement(store: SessionStore, a: InputAnnouncement): void {
  const outbox = (store.getStep(ANNOUNCE_OUTBOX) as InputAnnouncement[] | undefined) ?? [];
  store.delStep(ANNOUNCE_OUTBOX);
  store.putStep(ANNOUNCE_OUTBOX, [...outbox, a]);
}
function announcement(env: { agent: string; sessionId: string }, body: AnnouncementBody): InputAnnouncement {
  return { id: crypto.randomUUID(), agent: env.agent, session: env.sessionId, at: new Date().toISOString(), ...body };
}
// The proactive variant stands alone: it's the only trigger a CALLER may pass explicitly
// (TurnInput / ChannelContext.run / receive()) — inbound is derived from the event, resume is
// engine-internal. Narrowing the public seams to this type makes misuse (passing an inbound or
// resume trigger to the proactive entrypoint) unrepresentable, and keeps the /turn wire payload
// trivially JSON-serializable (no embedded event.raw).
export type ProactiveTrigger = { kind: "proactive"; by: string; note?: string }; // a schedule, another channel, the agent itself
export type TurnTrigger =
  | { kind: "inbound"; event: InboundEvent }          // a channel event (message, mention, reaction)
  | ProactiveTrigger
  | { kind: "resume"; callId: string };                // continuation after an input.requested suspend
export type TurnEvent =
  | { type: "turn.started"; turnId: string; trigger: TurnTrigger }
  | { type: "action.requested"; turnId: string; call: ToolCall }
  // `error`: the tool threw; `result` is `{ error }` and the model was told (#232).
  | { type: "action.completed"; turnId: string; call: ToolCall; result: unknown; error?: string }
  // A FileRef found in a tool's result (docs/rfc-agent-files.md), one event per file, right
  // after that call's action.completed. Structural: derived from the tool message in the log.
  | { type: "file.attached"; turnId: string; callId: string; file: FileRef }
  | { type: "message.completed"; turnId: string; text: string }
  | { type: "input.requested"; turnId: string; request: InputRequest }
  // The park of `turnId` ended (#260): answered by `by`, or retired by a session reset. Live
  // only; the durable record is the "resolved" InputAnnouncement.
  | { type: "input.resolved"; turnId: string; inputId: string; outcome: "answered" | "retired"; by?: string }
  | { type: "turn.completed"; turnId: string; text: string }
  | { type: "turn.failed"; turnId: string; error: TurnError; phase?: TurnFailurePhase; step?: string }
  // The turn was cancelled at a checkpoint boundary — no reply follows. `reason` says
  // why ("replaced" | "reset" | "requested") so a renderer can pick honest copy.
  // Live-only, like turn.failed: not foldable from the durable log.
  | { type: "turn.cancelled"; turnId: string; reason: CancelReason }
  | { type: "reasoning.delta"; turnId: string; text: string }
  | { type: "message.delta"; turnId: string; text: string };
export type TurnResult =
  | { status: "completed"; text: string }
  | { status: "suspended"; request: InputRequest }
  | { status: "failed"; error: TurnError }
  | { status: "cancelled" };

// A turn failure, serialized AT THE THROW SITE — the one place the real Error still
// exists. Everything downstream of the event sink (SSE across the worker→DO isolate,
// onTurnError, a channel render) sees only this shape, so whatever isn't captured here
// is gone: for a detached turn this is the ONLY failure-surfacing path (#96).
// `causeChain` walks `error.cause` outermost-first (messages only — one stack is
// enough to locate the site, the chain explains why).
export type TurnError = { message: string; stack?: string; causeChain?: string[] };
// Which engine step was in flight when the turn died — turns "fetch failed" into
// "the model call failed" vs "tool tool:3:call_7 failed". `step` on turn.failed carries
// the precise step id (`model:<n>` / `tool:<n>:<callId>`). Absent when the failure struck
// between steps (transcript reads, status writes).
export type TurnFailurePhase = "model" | "tool";

export function serializeTurnError(err: unknown): TurnError {
  if (err instanceof Error) {
    const causeChain: string[] = [];
    // depth-capped: a cyclic cause chain (err.cause === err) must not spin forever
    for (let c: unknown = err.cause, depth = 0; c !== undefined && depth < 8; depth++) {
      causeChain.push(c instanceof Error ? c.message : stringifyThrown(c));
      c = c instanceof Error ? c.cause : undefined;
    }
    return { message: err.message, ...(err.stack ? { stack: err.stack } : {}), ...(causeChain.length ? { causeChain } : {}) };
  }
  return { message: stringifyThrown(err) };
}
// Non-Error throwables keep their JSON shape instead of collapsing to "[object Object]".
function stringifyThrown(v: unknown): string {
  if (typeof v === "string") return v;
  try {
    const s = JSON.stringify(v);
    if (s !== undefined) return s;
  } catch { /* cyclic or hostile toJSON — fall through */ }
  return String(v);
}

// The event bus a turn emits to; observers (channels, an SSE surface) subscribe. Replaces
// the old coarse `Broadcaster.publish(turnId)` poke with typed events.
export interface EventSink {
  emit(e: TurnEvent): void;
  subscribe(cb: (e: TurnEvent) => void): () => void;
}

// Crash injection for the durability contract: throw at a chosen checkpoint
// boundary so a test can assert exactly-once across a real replay. Opt-in via
// `opts.crash`; absent in every production turn. Kept in the engine so the
// guarantee stays verifiable against the SAME code that ships.
export type Crash = {
  at: "before-model-commit" | "after-model-commit" | "before-tool-commit" | "after-tool-commit";
  step: string;
};

function assertCrash(crash: Crash | undefined, at: Crash["at"], step: string) {
  if (crash && crash.at === at && crash.step === step) throw new Error(`CRASH ${at} ${step}`);
}

// ── the engine: one durable turn ──────────────────────────────────────────────
// Reserved step key for the session's initiator principal (#128). It lives in the steps
// table beside the "suspended" checkpoint and the per-turn step ids ("model:N",
// "tool:N:<id>", "input:<turn>:<id>") — a namespace no turn-scoped key can collide with —
// so recording it needs no SessionStore contract change and it survives eviction with
// the rest of the durable log.
const INITIATOR_STEP = "session:initiator";

export async function runTurn(
  store: SessionStore,
  sink: EventSink,
  model: Model,
  tools: Tool[],
  opts: { turnId: string; userText: string; crash?: Crash; cancelled?: () => CancelReason | undefined },
  env: { runtime: Runtime; agent: string; sessionId: string; event?: InboundEvent; systemOverlay?: string; systemMode?: "append" | "replace"; deniedTools?: string[]; trigger?: TurnTrigger; announce?: boolean; files?: FileStore },
): Promise<string> {
  // Cancellation (#129) is polled at CHECKPOINT BOUNDARIES only — before the opening
  // commits, before each model call, between model deltas, and between tool calls — so a
  // cancelled turn always ends with the transcript in a state the next turn can build on.
  // `cancelled` returns the pending request's reason (or undefined): the reason rides the
  // signal and the turn.cancelled event so renderers can say WHY the reply never came.
  const cancelled = opts.cancelled ?? (() => undefined);
  // Superseded before it started (a queued turn a replace overtook): nothing has been
  // persisted for a fresh turn, so it vanishes without a trace. Emitted here — the loop's
  // catch handles in-flight cancellation, but this point is before the try.
  const early = cancelled();
  if (early) {
    sink.emit({ type: "turn.cancelled", turnId: opts.turnId, reason: early });
    throw new CancelSignal(opts.turnId, early);
  }
  // env.trigger overrides the derivation: a resume continuation announces { kind: "resume" }
  // even though it replays with the original inbound event.
  const trigger: TurnTrigger = env.trigger ?? (env.event ? { kind: "inbound", event: env.event } : { kind: "proactive", by: "system" });
  // An EXPLICITLY proactive turn (receive() / a schedule / another channel passed
  // env.trigger) opens with a `trigger` msg attributing who seeded it. A plain programmatic
  // turn (no event, no explicit trigger — CLI/http) is derived proactive/system for the live
  // turn.started event, but its opening stays a `user` msg: it's an API caller, not an
  // attributed agent-initiated seed. So key the role off the EXPLICIT trigger. See RFC §9 / #6.
  const opening: Msg | undefined = store.hasOpeningMessage(opts.turnId)
    ? undefined
    : env.trigger?.kind === "proactive"
      ? { role: "trigger", turnId: opts.turnId, text: opts.userText, by: env.trigger.by }
      : { role: "user", turnId: opts.turnId, text: opts.userText };
  // Durable session identity (#128): the FIRST resolved principal any turn arrives with
  // becomes the session's initiator — recorded once under a reserved step key (so the
  // SessionStore contract stays untouched and it survives eviction with the rest of the
  // log), immutable thereafter. The seat is claimed in the SAME transaction as the opening
  // message: committed separately, a crash could land between "session durably opened" and
  // "seat claimed", and on replay a later turn's different principal would take a seat that
  // was already spoken for. An anonymous opener still leaves the seat empty on purpose —
  // the first identified turn claims it then. The tx guard mirrors the suspend
  // checkpoint's: a crash-replay or a raced duplicate must not double-INSERT the step.
  let initiator = store.getStep(INITIATOR_STEP) as Principal | undefined;
  const claimant = initiator === undefined && env.event?.principal != null ? env.event.principal : undefined;
  // Notes still held (#262) — behind a batch a cancelled turn closed, say — join the log ahead
  // of this turn's opening, once the log is clean. Not for a parked turn's replay: its batch
  // is still owed, and the turn loop releases them once it is answered.
  const releaseNotes = store.getStep(HELD_NOTES) !== undefined && !owesToolResults(store.messages());
  if (opening !== undefined || claimant !== undefined || releaseNotes) {
    store.tx(() => {
      if (releaseNotes) releaseHeldNotes(store);
      if (opening !== undefined) store.appendMessage(opening);
      if (claimant !== undefined && store.getStep(INITIATOR_STEP) === undefined) store.putStep(INITIATOR_STEP, claimant);
    });
    // Read back rather than assume: if a raced duplicate won the guard, its principal holds the seat.
    if (claimant !== undefined) initiator = store.getStep(INITIATOR_STEP) as Principal | undefined;
  }
  store.setStatus("running");
  sink.emit({ type: "turn.started", turnId: opts.turnId, trigger });
  // A crash between a tool's commit and its file.attached emission left those files
  // unannounced; the replay's renderer is subscribed now, so announce them (see announceFiles).
  announcePendingFiles(store, sink, opts.turnId);

  // Principal gate: on a turn without a resolved trusted identity, requiresPrincipal
  // tools are absent by construction — not listed to the model, not resolvable by
  // toolStep (a hallucinated call fails loudly as an unknown tool). Filtered per-turn
  // (not at registration) because the same agent serves identified and anonymous
  // conversations; the filter is stable across a crash-replay since the principal
  // rides the persisted event.
  const principled = env.event?.principal != null ? tools : tools.filter((t) => !t.requiresPrincipal);
  // Source deny gate (#149): tools the turn's channel policy removes are absent the
  // same way — unlisted AND undispatchable — so per-surface policy is mechanical,
  // never persuaded in prose. Stable across crash-replay/resume: the deny list is
  // derived from the session's channel policies at start() and persisted with the
  // suspend checkpoint, like the overlay.
  const active = env.deniedTools?.length ? principled.filter((t) => !env.deniedTools!.includes(t.spec.name)) : principled;
  const specs = active.map((t) => t.spec);
  // Which durable step is in flight, for turn.failed attribution: set before each step
  // await, cleared on its return, so a throw OUTSIDE a step (transcript read, status
  // write) isn't blamed on the previously completed one.
  let inFlight: { phase: TurnFailurePhase; step: string } | undefined;
  try {
    while (true) {
      const msgs = store.messages();
      // Non-null: the loop always runs with ≥1 non-note message (the opening is appended
      // above before the first iteration). Notes (#262) are skipped: context, not a step.
      const last = lastNonNote(msgs)!;

      if (last.role === "assistant" && last.toolCalls.length === 0) {
        store.setStatus("done");
        sink.emit({ type: "turn.completed", turnId: opts.turnId, text: last.text });
        return last.text;
      }
      // The calls still owed a result: those of THIS turn's latest assistant message with
      // no answering tool message after it. Read off the transcript (not the step cache)
      // so a replay that crashed mid-batch resumes the remaining calls instead of asking
      // the model with a tool_use left dangling.
      const pending = unansweredCalls(msgs, opts.turnId);
      if (pending) {
        for (const call of pending.calls) {
          // Cancelled mid-batch: the calls already run stay committed; every call still
          // pending gets a synthetic "cancelled" result IN ONE TX, so no tool_use dangles
          // unanswered — the transcript stays valid for the next turn's model call (and
          // the model sees the work was interrupted, not that it silently vanished).
          const why = cancelled();
          if (why) {
            cancelToolBatch(store, pending.calls, pending.at, opts.turnId);
            throw new CancelSignal(opts.turnId, why);
          }
          const stepId = toolStepId(pending.at, call.id);
          inFlight = { phase: "tool", step: stepId };
          await toolStep(store, sink, active, call, stepId, opts, { ...env, initiator });
          inFlight = undefined;
        }
        continue;
      }
      // About to ask the model: the transcript is provider-clean here (every tool_use
      // answered). Notes held behind the batch just answered (#262) join the log now, so
      // the model reads them on this very call; the loop then re-reads the log.
      if (store.getStep(HELD_NOTES) !== undefined && store.tx(() => releaseHeldNotes(store)) > 0) continue;
      // A cancelled turn simply ends without a final assistant message.
      const why = cancelled();
      if (why) throw new CancelSignal(opts.turnId, why);
      inFlight = { phase: "model", step: `model:${msgs.length}` };
      await modelStep(store, sink, model, specs, `model:${msgs.length}`, msgs, opts, env.systemOverlay, env.systemMode, cancelled);
      inFlight = undefined;
    }
  } catch (err) {
    if (err instanceof CancelSignal) {
      // Cancelled at a boundary: committed steps stay, nothing dangles. Status returns to
      // "done" so the session is immediately ready for the replacement turn. No turn.failed
      // — cancellation is an outcome, not an error; result() maps it to { status: "cancelled" }.
      store.setStatus("done");
      sink.emit({ type: "turn.cancelled", turnId: opts.turnId, reason: err.reason });
      throw err;
    }
    if (err instanceof SuspendSignal) {
      // park the turn: persist everything a later resume needs to validate + replay (the pending
      // request + the turn's userText/overlay/event), mark suspended, and announce input.requested.
      // raw is stripped so the checkpoint stays JSON-serializable on the edge. Guarded put: a
      // redelivered replay of an already-parked turn re-announces WITHOUT re-inserting (putStep
      // is insert-only on the SQL stores).
      const event = env.event ? { ...env.event, raw: undefined } : undefined;
      const checkpoint: SuspendedCheckpoint = { turnId: opts.turnId, callId: err.callId, request: err.request, userText: opts.userText, systemOverlay: env.systemOverlay, systemMode: env.systemMode, deniedTools: env.deniedTools, event };
      store.tx(() => {
        if (store.getStep("suspended") === undefined) {
          store.putStep("suspended", checkpoint);
          if (env.announce) {
            const queued = heldBehind(store, opts.turnId);
            recordAnnouncement(store, announcement(env, { kind: "parked", turnId: opts.turnId, request: err.request, event, queued }));
          }
        }
        store.setStatus("suspended");
      });
      sink.emit({ type: "input.requested", turnId: opts.turnId, request: err.request });
      throw err; // AgentSession.result maps this to { status: "suspended" }
    }
    // includes intentional crash-injection throws (a failed ATTEMPT; replay re-runs).
    sink.emit({ type: "turn.failed", turnId: opts.turnId, error: serializeTurnError(err), phase: inFlight?.phase, step: inFlight?.step });
    throw err;
  }
}

// A tool step's durable id: the transcript index of the assistant message that made the
// call (the same n as its `model:<n>` step) plus the provider's call id. The index is
// what makes it unique — a provider id is only unique within one reply, and some
// providers mint "call_0" on every reply. Keyed by call id alone, a later call reusing
// an id found the earlier step cached, skipped the tool without answering it, and the
// turn loop spun on the unchanged transcript forever, synchronously (#167).
function toolStepId(assistantAt: number, callId: string): string {
  return `tool:${assistantAt}:${callId}`;
}

// ── notes held behind a tool batch (#262) ─────────────────────────────────────
// The Messages API requires a tool_use to be answered by the very next message: "You cannot
// include any messages between the assistant's tool use message and the user's tool result
// message." A note written while the log ends mid-batch — a turn parked on requestInput —
// would land exactly there. So it waits under this step and joins the log at the next point
// the log is clean: in the turn loop, once every call is answered and before the model is
// asked (the adapter then renders it after the tool results, in the same user message — the
// API allows text after tool_result blocks), or ahead of the next turn's opening. A reset
// archives held notes with the rest of the steps.
const HELD_NOTES = "held-notes";
type NoteMsg = Extract<Msg, { role: "note" }>;

// True when the log's latest assistant message made calls that are not all answered yet.
function owesToolResults(msgs: Msg[]): boolean {
  for (let at = msgs.length - 1; at >= 0; at--) {
    const m = msgs[at]!;
    if (m.role !== "assistant") continue;
    if (m.toolCalls.length === 0) return false;
    const answered = new Set<string>();
    for (let j = at + 1; j < msgs.length; j++) {
      const r = msgs[j]!;
      if (r.role === "tool") answered.add(r.toolCallId);
    }
    return m.toolCalls.some((c) => !answered.has(c.id));
  }
  return false;
}

// Move the held notes into the log, oldest first. Runs inside the caller's transaction;
// returns how many moved.
function releaseHeldNotes(store: SessionStore): number {
  const held = store.getStep(HELD_NOTES) as NoteMsg[] | undefined;
  if (!held?.length) return 0;
  for (const n of held) store.appendMessage(n);
  store.delStep(HELD_NOTES);
  return held.length;
}

// The latest message that is not a note — what the turn loop decides on. A note appended
// after a completed turn must not make a redelivery of that turn look unfinished.
function lastNonNote(msgs: Msg[]): Msg | undefined {
  for (let i = msgs.length - 1; i >= 0; i--) if (msgs[i]!.role !== "note") return msgs[i];
  return undefined;
}

// This turn's latest assistant message, if it made calls, and those of its calls that no
// tool message answers yet. Undefined when nothing is owed (no calls, all answered, or
// the latest assistant message belongs to an earlier turn).
function unansweredCalls(msgs: Msg[], turnId: string): { at: number; calls: ToolCall[] } | undefined {
  for (let at = msgs.length - 1; at >= 0; at--) {
    const m = msgs[at]!;
    if (m.role !== "assistant") continue;
    if (m.turnId !== turnId || m.toolCalls.length === 0) return undefined;
    const answered = new Set<string>();
    for (let j = at + 1; j < msgs.length; j++) {
      const r = msgs[j]!;
      if (r.role === "tool") answered.add(r.toolCallId);
    }
    const calls = m.toolCalls.filter((c) => !answered.has(c.id));
    return calls.length ? { at, calls } : undefined;
  }
  return undefined;
}

// Close a cancelled turn's partially-run tool batch: every call WITHOUT a committed step
// gets a synthetic "cancelled" result (step + tool message in one tx), so no tool_use is
// left unanswered — providers reject a transcript with a dangling call, and the next
// turn's model call must be able to build on this one. No action.completed is emitted for
// a synthetic result: nothing ran, and turn.cancelled follows immediately.
function cancelToolBatch(store: SessionStore, calls: ToolCall[], assistantAt: number, turnId: string) {
  store.tx(() => {
    for (const call of calls) {
      const stepId = toolStepId(assistantAt, call.id);
      if (store.getStep(stepId) !== undefined) continue;
      const result = { cancelled: true, reason: "the turn was cancelled before this call ran" };
      store.putStep(stepId, result);
      store.appendMessage({ role: "tool", turnId, toolCallId: call.id, name: call.name, result });
    }
  });
}

async function modelStep(
  store: SessionStore,
  sink: EventSink,
  model: Model,
  specs: ToolSpec[],
  stepId: string,
  msgs: Msg[],
  opts: { turnId: string; crash?: Crash },
  systemOverlay?: string,
  systemMode?: "append" | "replace",
  cancelled?: () => CancelReason | undefined,
) {
  if (store.getStep(stepId) !== undefined) return; // cached: assistant already appended in the same tx
  // Iterate the model stream: forward reasoning/answer tokens as LIVE TurnEvents (not
  // persisted, not re-emitted on replay), and take the terminal `done.reply` as the value
  // to checkpoint. The adapter's `done` is authoritative — the engine never assembles the
  // reply from partial text/tool deltas.
  let reply: ModelReply | undefined;
  let finish: ModelFinish | undefined;
  for await (const d of model(msgs, specs, systemOverlay !== undefined ? { system: systemOverlay, ...(systemMode ? { systemMode } : {}) } : undefined)) {
    // Cancelled mid-stream: nothing persisted for this step, so throwing here (which
    // closes the provider iterator via for-await's return()) just discards the partial
    // reply — the checkpoint machine treats it like a model call that never happened.
    // Polling per delta means cancellation lands without a Model contract change; a
    // provider that is silent between deltas is only interrupted at its next token.
    const why = cancelled?.();
    if (why) throw new CancelSignal(opts.turnId, why);
    if (d.type === "reasoning") sink.emit({ type: "reasoning.delta", turnId: opts.turnId, text: d.text });
    else if (d.type === "text") sink.emit({ type: "message.delta", turnId: opts.turnId, text: d.text });
    else { reply = d.reply; finish = d.finish; break; } // `done` is terminal: `break` cancels the iterator (return()) so
    // extra deltas / a throw AFTER the authoritative reply can't turn a completed turn into a failure
  }
  if (!reply) throw new Error("model stream ended without a `done` event");
  // An ABNORMAL stop with an EMPTY reply fails the step instead of committing: providers
  // signal truncation/filtering via the finish reason, and both major APIs document that
  // such stops may carry no content at all. Committing "" here would let the turn
  // "complete" silently — a channel then renders nothing and nobody is told why. A reply
  // WITH content under an abnormal reason still commits (truncated-but-usable), and an
  // adapter that reports no finish keeps today's behavior — an empty completion stays
  // legitimate where it is deliberate (tool-only turns end with empty text by design).
  // Thrown BEFORE the checkpoint, so a retry re-runs the model like any model failure.
  if (finish && finish.reason !== "stop" && !reply.text.trim() && reply.toolCalls.length === 0) {
    const detail = finish.raw && finish.raw !== finish.reason ? ` (provider: ${finish.raw})` : "";
    throw new Error(`model stopped abnormally — ${finish.reason}${detail} — and returned an empty reply`);
  }
  // Recheck before committing: `break` on the done delta awaits the provider iterator's
  // return() (cleanup can yield), so a cancellation landing during that await would
  // otherwise commit the stale reply — and a call-less reply would then COMPLETE the
  // superseded turn. Nothing is persisted yet, so cancelling here is still free.
  const lateCancel = cancelled?.();
  if (lateCancel) throw new CancelSignal(opts.turnId, lateCancel);
  assertCrash(opts.crash, "before-model-commit", stepId); // nothing persisted → replay re-asks the model
  store.tx(() => {
    store.putStep(stepId, reply);
    store.appendMessage({ role: "assistant", turnId: opts.turnId, text: reply.text, toolCalls: reply.toolCalls });
  });
  assertCrash(opts.crash, "after-model-commit", stepId); // committed → replay skips (exactly-once append)
  // an assistant message finalized (text when present) + one action.requested per tool call
  if (reply.text.trim()) sink.emit({ type: "message.completed", turnId: opts.turnId, text: reply.text });
  for (const call of reply.toolCalls) sink.emit({ type: "action.requested", turnId: opts.turnId, call });
}

async function toolStep(
  store: SessionStore,
  sink: EventSink,
  tools: Tool[],
  call: ToolCall,
  stepId: string,
  opts: { turnId: string; crash?: Crash },
  env: { runtime: Runtime; agent: string; sessionId: string; event?: InboundEvent; initiator?: Principal; files?: FileStore },
) {
  if (store.getStep(stepId) !== undefined) return;
  const tool = tools.find((t) => t.spec.name === call.name);
  if (!tool) throw new Error(`unknown tool ${call.name}`);
  const remote = toolMode(tool.run, tool.mode) === "remote";
  const ctx: ToolContext = {
    store, runtime: env.runtime, agent: env.agent, sessionId: env.sessionId, callId: call.id, event: env.event, principal: env.event?.principal, initiator: env.initiator,
    // replay-aware: return the stored answer if resume already provided it, else park the turn.
    // Answers are TURN-scoped (`input:${turnId}:${id}`): a later turn re-asking the same id must
    // park again — never silently reuse a prior turn's answer (an approval must not carry over).
    // Deliberately NOT an async function: a sync (local) tool that misuses it must fail loudly
    // and synchronously — an async requestInput would hand it a rejected Promise it can't await,
    // which the local-tool tx would then commit as its checkpointed result.
    requestInput: (req) => {
      if (!remote) throw new FatalToolError(`requestInput: tool "${call.name}" runs local — only a remote tool (an async run, or mode: "remote") can park the turn awaiting input`);
      const answer = store.getStep(`input:${opts.turnId}:${req.id}`);
      if (answer !== undefined) return Promise.resolve((answer as { input: unknown }).input);
      throw new SuspendSignal({ id: req.id, prompt: req.prompt, schema: req.schema, answerers: req.answerers ?? defaultAnswerers(env.event, req.id) }, call.id);
    },
    files: env.files ?? NO_FILE_STORE,
  };

  assertCrash(opts.crash, "before-tool-commit", stepId); // nothing done → safe clean re-run
  const record = (result: unknown, error: string | undefined) => {
    store.putStep(stepId, result);
    store.appendMessage({ role: "tool", turnId: opts.turnId, toolCallId: call.id, name: call.name, result, ...(error !== undefined ? { isError: true as const } : {}) });
  };

  let out: unknown;
  let error: string | undefined;
  // Only what the TOOL throws becomes its result. A failure recording a successful result
  // (the store's own putStep / appendMessage) is infrastructure and must still fail the
  // turn — so the local path marks a tool throw before rethrowing it out of the tx (the
  // rethrow is what rolls the tx back).
  let toolThrew = false;
  try {
    if (remote) {
      // network / subagent side effect: at-least-once (can't 2PC with local storage;
      // a subagent is itself durable, and its child turnId makes replay idempotent)
      toolThrew = true;
      out = await tool.run(call.input, ctx);
      toolThrew = false;
    } else {
      // local side effect: exactly-once (side effect + checkpoint + append in ONE tx)
      store.tx(() => {
        toolThrew = true;
        out = tool.run(call.input, ctx);
        toolThrew = false;
        // A Promise here means the tool is async but was classified local (#233). Committing it
        // would record `{}` and lose the real result, so fail the turn — it's a definition bug.
        // The Promise already runs; swallow its rejection so it can't surface as unhandled.
        if (isThenable(out)) {
          out.then(undefined, () => {});
          throw new FatalToolError(`tool "${call.name}" returned a Promise but runs local (sync) — declare its run \`async\`, or set mode: "remote"`);
        }
        record(out, undefined);
      });
    }
  } catch (err) {
    // The runtime's own control flow keeps propagating — park, cancel, a tool that declared
    // its error fatal — and so does anything that didn't come from the tool itself.
    if (!toolThrew || err instanceof SuspendSignal || err instanceof CancelSignal || err instanceof FatalToolError) throw err;
    // Anything else is the tool's RESULT (#232): the model sees the failure on its next step
    // and can react, instead of the whole turn dying. A local tool's tx rolled back, so its
    // side effect is undone and the error is recorded in a fresh tx below. Checkpointed like
    // any result, so a replay doesn't re-run the failed call.
    error = toolErrorText(err);
    out = { error };
  }
  if (remote || error !== undefined) store.tx(() => record(out, error));
  assertCrash(opts.crash, "after-tool-commit", stepId); // committed → replay skips (no duplicate side effect)
  sink.emit({ type: "action.completed", turnId: opts.turnId, call, result: out, ...(error !== undefined ? { error } : {}) });
  if (error === undefined) announceFiles(store, sink, stepId, opts.turnId, call.id, out);
}

// Emit file.attached for a tool result's FileRefs, then mark the step announced. Delivery
// keys off these LIVE events, and the marker is what tells a replay whether they went out:
//   • a crash after the tool committed but before this ran leaves no marker, so the replay
//     announces them (announcePendingFiles) — otherwise the files would never be delivered;
//   • a resume after a park finds the marker on every earlier step, so files already
//     delivered above the approval prompt are not uploaded again.
// At-least-once: a crash between the emission and the marker announces them twice.
function announceFiles(store: SessionStore, sink: EventSink, stepId: string, turnId: string, callId: string, result: unknown): void {
  const files = fileRefsIn(result);
  if (files.length === 0) return;
  const marker = `files:${stepId}`;
  if (store.getStep(marker) !== undefined) return;
  for (const file of files) sink.emit({ type: "file.attached", turnId, callId, file });
  store.tx(() => {
    if (store.getStep(marker) === undefined) store.putStep(marker, true); // putStep is insert-only
  });
}

// The turn's committed, successful tool results whose files were never announced — the
// crash window above. A tool message's step id is rebuilt the way toolStep built it: the
// index of the assistant message that made the call, plus the call id.
function announcePendingFiles(store: SessionStore, sink: EventSink, turnId: string): void {
  const msgs = store.messages();
  for (let i = 0; i < msgs.length; i++) {
    const m = msgs[i]!;
    if (m.role !== "tool" || m.turnId !== turnId || m.isError) continue;
    let at = i - 1;
    while (at >= 0 && !(msgs[at]!.role === "assistant" && (msgs[at] as Extract<Msg, { role: "assistant" }>).toolCalls.some((c) => c.id === m.toolCallId))) at--;
    if (at < 0) continue;
    announceFiles(store, sink, toolStepId(at, m.toolCallId), turnId, m.toolCallId, m.result);
  }
}

// The store a tool gets when its host installed none: storing a file is a deployment
// misconfiguration the model can't fix, so put() fails the turn instead of handing the model
// an error to retry. get() finds nothing, like any store asked for a file it doesn't hold.
const NO_FILE_STORE: FileStore = {
  put: async () => {
    throw new FatalToolError("ctx.files: this agent's host has no file store — configure a blob resource (resources.blob) or pass `files` to the runtime");
  },
  get: async () => null,
};

// ── transcript fold (pure; used by observe/transcript surfaces) ───────────────
export type Turn = {
  turnId: string;
  user: string;
  by?: string; // set when the turn was agent-initiated: who seeded it (the trigger msg's `by`)
  steps: { name: string; done: boolean; result?: unknown }[];
  text?: string;
  // Set on the entry a note (#262) folds into — its own, keyed by the note's id: a note is
  // not part of any turn. `user` is empty and `steps` stays empty on such an entry.
  note?: { by: string; kind: NoteKind; text: string; at: string };
};
export function foldTranscript(msgs: Msg[]): Turn[] {
  const byId = new Map<string, Turn>();
  const order: string[] = [];
  for (const m of msgs) {
    let t = byId.get(m.turnId);
    if (!t) { t = { turnId: m.turnId, user: "", steps: [] }; byId.set(m.turnId, t); order.push(m.turnId); }
    if (m.role === "user") t.user = m.text;
    else if (m.role === "note") t.note = { by: m.by, kind: m.kind, text: m.text, at: m.at };
    else if (m.role === "trigger") { t.user = m.text; t.by = m.by; } // proactive seed shows as the turn's prompt, still attributed
    else if (m.role === "assistant") {
      for (const tc of m.toolCalls) t.steps.push({ name: tc.name, done: false });
      if (m.text) t.text = m.text;
    } else if (m.role === "tool") {
      const s = t.steps.find((x) => x.name === m.name && !x.done) ?? t.steps[t.steps.length - 1];
      if (s) { s.done = true; s.result = m.result; }
    }
  }
  return order.map((id) => byId.get(id)!);
}

// ── turn id minting (#95) ─────────────────────────────────────────────────────
// Globally unique, lexically time-sortable: `t_` + a monotonic ULID (48-bit ms
// timestamp + 80-bit randomness, Crockford base32). Replaces the per-actor
// sequence (`t1`, `t2`, …), which collided in BOTH dimensions: across sessions
// (every session's first turn was "t1" — useless in any table keyed by turnId),
// and within one session across a DO hibernation (the in-memory seq reset to 0,
// re-minting "t1" — which hasOpeningMessage then treated as a REDELIVERY of the
// old turn, silently replaying its steps instead of starting a new turn).
// The `t_` prefix keeps a mixed ledger ordered across the migration boundary:
// every legacy id ("t" + digits — "t1", "t42", …) sorts before every new id
// ("t_01…"), because the character after "t" decides: any digit "0"–"9" < "_".
// Monotonic within one isolate: same-ms mints increment the random block, so a
// ledger ordered by id stays insertion-ordered even inside a burst.
const B32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"; // Crockford
let lastMs = -1;
const lastRand = new Uint8Array(10);
export function mintTurnId(): string {
  const now = Date.now();
  if (now > lastMs) {
    lastMs = now;
    crypto.getRandomValues(lastRand);
  } else {
    // same (or clock-regressed) ms: +1 with carry keeps ids strictly increasing
    let i = 9;
    for (; i >= 0; i--) { if (++lastRand[i]! <= 0xff) break; lastRand[i] = 0; }
    // full 80-bit rollover (a random start adjacent to the max, or a clock stuck
    // behind): borrow into the time block so the invariant stays absolute
    if (i < 0) lastMs++;
  }
  let time = "";
  for (let ms = lastMs, i = 0; i < 10; i++) { time = B32[ms % 32]! + time; ms = Math.floor(ms / 32); }
  let rand = "";
  for (let i = 0; i < 10; i += 5) {
    // 5 bytes → 8 base32 chars, MSB-first
    let acc = 0;
    for (let j = 0; j < 5; j++) acc = acc * 256 + lastRand[i + j]!;
    let block = "";
    for (let j = 0; j < 8; j++) { block = B32[acc % 32]! + block; acc = Math.floor(acc / 32); }
    rand += block;
  }
  return `t_${time}${rand}`;
}

// ── the outer seam: AgentSession actor (serializes turns via an inbox) ─────────
// `trigger` marks an agent-INITIATED turn (proactive): no inbound event, the turn opens with a
// `trigger`-role seed attributed to `by` (a schedule, another channel, the agent). Omitted for a
// normal inbound/programmatic turn. Proactive-only by type: inbound is derived from `event`,
// resume is engine-internal. See §9 / receive().
// `replace` (#129) is the debounce affordance: before this turn is queued, request
// cancellation of every unfinished turn on the session (in-flight and queued) — they
// unwind at their next checkpoint boundary and this turn runs after them on the chain.
// A SUSPENDED turn is deliberately NOT cancelled: a parked approval must not silently
// die because a new message arrived.
// `ifSuspended` (#263) says what an inbound turn does against a parked session: "reject"
// (the default — right for an interactive surface, where the person who would speak next is
// the one being asked) or "queue" (right where the other party keeps talking regardless, like
// email): the turn is held durably and runs, in arrival order, once the park resolves. See
// QueuedTurn for `hostContext`.
export type TurnInput = { turnId?: string; userText: string; crash?: Crash; event?: InboundEvent; trigger?: ProactiveTrigger; replace?: boolean; ifSuspended?: "reject" | "queue"; hostContext?: unknown };

export class AgentSession {
  private chain: Promise<unknown> = Promise.resolve();
  // Per-turn terminal promise, so result(turnId) can await a turn started with start().
  private readonly running = new Map<string, Promise<string>>();
  // Turn ids whose cancellation was requested (#129), each with WHY. In-memory on
  // purpose: cancellation is best-effort liveness (the engine polls at checkpoint
  // boundaries), not a durability contract — after a crash the request is gone and a
  // replay runs uncancelled.
  private readonly cancelRequests = new Map<string, CancelReason>();
  // A reset() whose archival op is still queued on the chain (#129). While set, start()
  // lets a new turn queue PAST a suspended park (the reset will retire it first) and
  // resume() refuses (the park is being retired — answering it would race the archival).
  private pendingReset = false;
  // Called when a held inbound turn (#263) is started — synchronously, right after start(),
  // before any of its events can emit — so a host can attach what the turn's original caller
  // would have (a delivered render). Set by the host after constructing the session; queued
  // turns started without one simply run.
  onDequeue?: (turnId: string, queued: QueuedTurn) => void;
  // The host's receiver for InputAnnouncements (#260). Announcements are recorded only while
  // one is set; the host sets it after constructing the session and then calls
  // flushAnnouncements() to deliver what an earlier life left undelivered.
  onAnnounce?: (a: InputAnnouncement) => void | Promise<void>;
  // Where this session's tools store the files they produce (ctx.files) — set by the host
  // after constructing the session, like onAnnounce. Absent: a tool's ctx.files.put fails
  // the turn naming the missing configuration.
  files?: FileStore;
  private flushing?: Promise<void>;
  private flushAgain = false;
  // Explicit fields + assignment (not constructor parameter properties): June
  // ships raw .ts, so consumers type-strip it — parameter properties aren't
  // erasable and break `erasableSyntaxOnly` / Node native strip-types.
  readonly agent: string;
  readonly id: string;
  private readonly store: SessionStore;
  private readonly sink: EventSink;
  private readonly model: Model;
  private readonly tools: Tool[];
  private readonly runtime: Runtime;
  // Per-source turn policies: when a turn's InboundEvent.source matches a key, that
  // policy applies — an overlay appended to (or, under mode "replace", replacing) the
  // system prompt, and/or a mechanical tool-deny list (see ChannelPolicy). A bare
  // string is the classic overlay-only form. Lets ONE shared agent branch its
  // behavior by real, unforgeable channel source — no userText markers.
  private readonly channelInstructions?: Record<string, string | ChannelPolicy>;
  constructor(agent: string, id: string, store: SessionStore, sink: EventSink, model: Model, tools: Tool[], runtime: Runtime, channelInstructions?: Record<string, string | ChannelPolicy>) {
    this.agent = agent;
    this.id = id;
    this.store = store;
    this.sink = sink;
    this.model = model;
    this.tools = tools;
    this.runtime = runtime;
    this.channelInstructions = channelInstructions;
  }

  // Kick a turn off WITHOUT awaiting it — returns its id so a caller can `observe` the
  // live event stream and `result(turnId)` for the terminal state. This is the primary
  // interface for live/streaming consumers (channels, SSE). Turns are serialized: each
  // awaits the previous on the chain — no interleaving on the shared transcript. (On a
  // Durable Object this is blockConcurrencyWhile; here it's a promise chain.)
  start(input: TurnInput): { turnId: string; queued?: true } {
    const turnId = input.turnId ?? mintTurnId();
    // `replace` supersedes unfinished turns; for held turns that would mean silently dropping
    // messages someone sent and was told were accepted. The combination is refused.
    if (input.ifSuspended === "queue" && input.replace) {
      throw new Error(`start(): ifSuspended "queue" and replace cannot be combined — replacing would drop held turns that were accepted`);
    }
    // HOLD (#263): against a parked session — or behind turns already held, so arrival order
    // holds — an inbound turn that asked to queue is recorded durably instead of rejected.
    // Never the parked turn itself: its redelivery replays and re-parks, as it always has
    // (holding it would run it a second time behind the held turns).
    const parked = this.parkedTurnId();
    if (input.ifSuspended === "queue" && !this.pendingReset && turnId !== parked && (parked !== undefined || this.heldTurns().length > 0)) {
      this.hold(turnId, input);
      this.drain(); // no-op while parked; runs the head if the park resolved in between
      return { turnId, queued: true };
    }
    // A turn that runs now still goes after turns a restart left held once their park
    // resolved (a no-op in every other case) — so held turns keep their place in line.
    if (parked === undefined) this.drain();
    return this.begin(turnId, input);
  }

  // Schedule a turn on the chain. start() decides whether it runs now; drain() starts held
  // turns through here directly, so starting one cannot re-enter drain() and start the next.
  private begin(turnId: string, input: TurnInput): { turnId: string } {
    // While a turn is parked awaiting input, the transcript ends in its dangling tool call —
    // running a NEW turn on it would corrupt both (the resumed replay would adopt the new
    // turn's tail as its own result). Reject loudly; redelivering the SAME parked turn is
    // allowed (it replays, re-parks, and re-announces input.requested).
    // A PENDING RESET waives this check: the queued archival retires the park before this
    // turn runs (chain order), so the turn lands on a fresh transcript — exactly the
    // "reset and start over" sequence. The run-time guard below still protects the park
    // if that archival fails.
    if (this.store.getStatus() === "suspended" && !this.pendingReset) {
      const s = this.store.getStep("suspended") as SuspendedCheckpoint | undefined;
      if (s && s.turnId !== turnId) {
        throw new Error(`session is suspended awaiting input "${s.request.id}" (turn ${s.turnId}); resume it before starting a new turn`);
      }
    }
    // The replace affordance (#129): supersede every unfinished turn BEFORE queuing this
    // one — they unwind at their next checkpoint boundary, then this turn runs. Ordered
    // after the suspended check on purpose: replace never cancels a parked approval.
    if (input.replace) this.cancelPending("replaced");
    const rawPolicy = input.event ? this.channelInstructions?.[input.event.source] : undefined;
    const policy: ChannelPolicy | undefined = typeof rawPolicy === "string" ? { overlay: rawPolicy } : rawPolicy;
    const run = () => {
      // Run-time re-check of the park: the start()-time check can be stale by the time the
      // chain reaches this turn — an EARLIER queued turn may have parked meanwhile, or a
      // pending reset's archival may have FAILED (leaving the park it was meant to retire).
      // Running over a foreign park would corrupt it (the resumed replay would adopt this
      // turn's tail), so fail loudly instead; redelivering the SAME parked turn stays allowed.
      if (this.store.getStatus() === "suspended") {
        const s = this.store.getStep("suspended") as SuspendedCheckpoint | undefined;
        if (s && s.turnId !== turnId) {
          // (announced as turn.failed by withTerminal — runTurn never ran to do it)
          return Promise.reject(new Error(`session is suspended awaiting input "${s.request.id}" (turn ${s.turnId}); resume it before starting a new turn`));
        }
      }
      return runTurn(
        this.store,
        this.sink,
        this.model,
        this.tools,
        { turnId, userText: input.userText, crash: input.crash, cancelled: () => this.cancelRequests.get(turnId) },
        { runtime: this.runtime, agent: this.agent, sessionId: this.id, event: input.event, systemOverlay: policy?.overlay, systemMode: policy?.overlayMode, deniedTools: policy?.denyTools, trigger: input.trigger, announce: !!this.onAnnounce, files: this.files },
      );
    };
    this.track(turnId, this.chain.then(() => this.withTerminal(turnId, run)));
    return { turnId };
  }

  // Request cancellation of one unfinished turn (#129). It unwinds at its next checkpoint
  // boundary (between model deltas / tool calls) and settles as { status: "cancelled" }.
  // Returns false when the turn isn't running or queued here — already finished, parked
  // (a suspended turn is not cancellable), or never started.
  cancel(turnId: string): boolean {
    if (!this.running.has(turnId)) return false;
    this.cancelRequests.set(turnId, "requested");
    return true;
  }

  // The replace primitive: request cancellation of EVERY unfinished turn (in-flight and
  // queued). Returns the turn ids marked. A suspended turn is not in `running`, so a
  // parked approval survives — by design (see TurnInput.replace). `reason` rides the
  // CancelSignal and the turn.cancelled event ("replaced" from a replace turn, "reset"
  // from reset(), "requested" for an explicit call).
  cancelPending(reason: CancelReason = "requested"): string[] {
    const ids = [...this.running.keys()];
    for (const id of ids) this.cancelRequests.set(id, reason);
    return ids;
  }

  // Register the in-flight promise for result() and prune it on settle. The delete is tied to
  // THIS promise's identity: a suspend→resume reuses the same turnId, so the parked promise's
  // late cleanup must NOT clear the continuation's entry (delete by turnId alone would). Both
  // branches run cleanup, so the rejection is handled here — no unhandled rejection.
  // Every turn that ends in a rejection announces a terminal event. Live consumers (a host's
  // runStream, the DO's SSE stream) end on one — turn.completed / failed / cancelled or
  // input.requested — and wait forever without it. runTurn emits them itself, except where
  // it throws before or while emitting: the run-time park refusal above (runTurn never
  // runs), a store error while opening the turn, or one while recording a cancel or a
  // park. Those are announced here as turn.failed; a turn that already emitted its
  // terminal event is left alone. result() is unaffected — it reads the rejection.
  private withTerminal(turnId: string, run: () => Promise<string>): Promise<string> {
    let announced = false;
    const off = this.sink.subscribe((e) => {
      if (e.turnId === turnId && (e.type === "turn.completed" || e.type === "turn.failed" || e.type === "turn.cancelled" || e.type === "input.requested")) announced = true;
    });
    let p: Promise<string>;
    try { p = run(); } catch (err) { p = Promise.reject(err); }
    return p.then(
      (text) => { off(); return text; },
      (err: unknown) => {
        off();
        if (!announced) this.sink.emit({ type: "turn.failed", turnId, error: serializeTurnError(err) });
        throw err;
      },
    );
  }

  private track(turnId: string, p: Promise<string>): void {
    this.chain = p.catch(() => {}); // a failed turn must not break the inbox
    this.running.set(turnId, p);
    // The cancel request is cleared under the same identity guard: if a resume continuation
    // already replaced this entry, a pending request now belongs to the continuation.
    const clear = () => { if (this.running.get(turnId) === p) { this.running.delete(turnId); this.cancelRequests.delete(turnId); } };
    // Every settle is a chance to run a held turn (#263): the park may have resolved. A drain
    // that throws (a store error) must not become an unhandled rejection of this bookkeeping.
    const settle = () => {
      clear();
      void this.flushAnnouncements();
      try {
        this.release(turnId);
        this.drain();
      } catch (err) {
        console.error(`[june] agent "${this.agent}" session "${this.id}": starting a held turn failed:`, err);
      }
    };
    p.then(settle, settle);
  }

  // Await a turn's terminal state (completed | suspended | failed). Reads the in-flight promise
  // from start(); if the turn already finished/parked, falls back to the durable log (status +
  // the suspended checkpoint / final assistant text) so a late result() call still resolves.
  async result(turnId: string): Promise<TurnResult> {
    const p = this.running.get(turnId);
    if (p) {
      try { return { status: "completed", text: await p }; }
      catch (err) {
        if (err instanceof SuspendSignal) return { status: "suspended", request: err.request };
        if (err instanceof CancelSignal) return { status: "cancelled" };
        return { status: "failed", error: serializeTurnError(err) };
      }
    }
    if (this.store.getStatus() === "suspended") {
      const s = this.store.getStep("suspended") as SuspendedCheckpoint | undefined;
      if (s && s.turnId === turnId) return { status: "suspended", request: s.request }; // only THIS turn's park
    }
    const msgs = this.store.messages().filter((m) => m.turnId === turnId);
    const lastAssistant = [...msgs].reverse().find((m): m is Extract<Msg, { role: "assistant" }> => m.role === "assistant");
    if (lastAssistant && lastAssistant.toolCalls.length === 0) return { status: "completed", text: lastAssistant.text };
    return { status: "failed", error: { message: `no terminal result for turn ${turnId}` } };
  }

  // Provide the input a suspended turn is waiting on and continue it. Stores the answer as the
  // `input:{turnId}:{inputId}` checkpoint, then replays the turn: cached steps skip,
  // ctx.requestInput now finds its answer and returns, and the turn runs on (may complete, or
  // suspend again). `opts.by` is the resumer's VERIFIED identity — the caller must authenticate
  // it (e.g. the user id from a signed Slack interaction payload), never trust a client-supplied
  // value. `opts.granted` is a host's grant for a { policy } answerer (grantAnswer). When the
  // request names answerers, resume is DEFAULT-DENY: a { user } needs that exact `by`, a
  // { policy } needs a grant for exactly that policy and scope. Returns { turnId } like
  // start(); await result(turnId) for the new terminal state.
  resume(turnId: string, inputId: string, input: unknown, opts?: { by?: string; granted?: Answerers }): { turnId: string } {
    // A pending reset is about to retire the park this answer targets — applying it would
    // race the archival (the continuation could replay over an emptied store). Refuse.
    if (this.pendingReset) throw new Error(`turn ${turnId} cannot be resumed: a session reset is pending and will retire the suspended turn`);
    const suspended = this.store.getStep("suspended") as SuspendedCheckpoint | undefined;
    if (!suspended || suspended.turnId !== turnId) throw new Error(`turn ${turnId} is not suspended`);
    if (inputId !== suspended.request.id) {
      throw new Error(`turn ${turnId} is awaiting input "${suspended.request.id}", not "${inputId}"`);
    }
    const answerers = suspended.request.answerers;
    if (answerers) {
      const allowed = "user" in answerers
        ? opts?.by !== undefined && opts.by === answerers.user
        : opts?.granted !== undefined && sameAnswerers(opts.granted, answerers);
      if (!allowed) {
        throw new ResumeAuthorizationError(`resume: ${opts?.by ?? "<unidentified>"} is not authorized to answer input "${inputId}"`);
      }
    }
    this.store.tx(() => {
      this.store.putStep(`input:${turnId}:${inputId}`, { input });
      this.store.delStep("suspended"); // consumed — the next park (same turn or a later one) inserts cleanly
      this.store.setStatus("running");
      if (this.onAnnounce) recordAnnouncement(this.store, announcement({ agent: this.agent, sessionId: this.id }, { kind: "resolved", turnId, inputId, outcome: "answered", by: opts?.by }));
    });
    this.sink.emit({ type: "input.resolved", turnId, inputId, outcome: "answered", by: opts?.by });
    void this.flushAnnouncements();
    const run = () =>
      runTurn(
        this.store,
        this.sink,
        this.model,
        this.tools,
        { turnId, userText: suspended.userText, cancelled: () => this.cancelRequests.get(turnId) },
        { runtime: this.runtime, agent: this.agent, sessionId: this.id, event: suspended.event, systemOverlay: suspended.systemOverlay, systemMode: suspended.systemMode, deniedTools: suspended.deniedTools, trigger: { kind: "resume", callId: suspended.callId }, announce: !!this.onAnnounce, files: this.files },
      );
    this.track(turnId, this.chain.then(() => this.withTerminal(turnId, run)));
    return { turnId };
  }

  // Append an attributed note (#262) to the session's history without running a turn — an
  // operator's reply during a take-over, traffic a channel observed. The next model call reads
  // it as labelled context. Serialized with turns (queued behind whatever is running), allowed
  // while idle or parked. While a turn is parked its tool batch is still owed, so the note is
  // held and joins the log as soon as the batch is answered — before the resumed turn next
  // asks the model — keeping the Messages API's rule that nothing sits between a tool_use and
  // its tool_result. Resolves to the note's id once it is durably recorded (in the log, or held).
  note(input: { by: string; kind: NoteKind; text: string }): Promise<{ noteId: string }> {
    for (const field of ["by", "kind", "text"] as const) {
      if (typeof input[field] !== "string" || input[field].length === 0) {
        return Promise.reject(new TypeError(`note(): "${field}" must be a non-empty string`));
      }
    }
    const note: NoteMsg = { role: "note", turnId: `n_${mintTurnId().slice(2)}`, by: input.by, kind: input.kind, text: input.text, at: new Date().toISOString() };
    const store = this.store;
    const op = this.chain.then(() => {
      store.tx(() => {
        if (!owesToolResults(store.messages())) {
          store.appendMessage(note);
          return;
        }
        const held = (store.getStep(HELD_NOTES) as NoteMsg[] | undefined) ?? [];
        store.delStep(HELD_NOTES); // putStep is insert-only: replace
        store.putStep(HELD_NOTES, [...held, note]);
      });
      return { noteId: note.turnId };
    });
    this.chain = op.catch(() => {}); // a failed note must not break the inbox
    return op;
  }

  // The input this session is parked on, if any — what a host reads to evaluate a { policy }
  // answerer before the synchronous resume (grantAnswer).
  // `queued` counts the inbound turns held since the park (#263) — an approver's cue that the
  // conversation moved on after the request was made.
  pending(): { turnId: string; request: InputRequest; queued: number } | undefined {
    const suspended = this.store.getStep("suspended") as SuspendedCheckpoint | undefined;
    return suspended ? { turnId: suspended.turnId, request: suspended.request, queued: heldBehind(this.store, suspended.turnId) } : undefined;
  }

  // The inbound turns held while the session was parked (#263), oldest first.
  heldTurns(): QueuedTurn[] {
    return (this.store.getStep(INBOUND_QUEUE) as QueuedTurn[] | undefined) ?? [];
  }

  // Start the oldest held turn, if the session can run it now: no park, no turn running or
  // queued on the chain, no reset pending. One at a time — the next is drained when this one
  // settles (track), because a turn that parks again must hold everything behind it. Hosts
  // call it after rebuilding a session (a restart between a park resolving and its held turns
  // starting leaves them in the store); every other drain is automatic.
  // The entry stays held while its turn runs and is released only when the turn settles
  // (release) — so a crash anywhere in between leaves it held, and the next drain starts the
  // same turnId again, which replays from its checkpoints (an opening already recorded is not
  // recorded twice). Removing it here instead would lose the turn on a crash before its
  // opening is committed.
  drain(): { turnId: string } | undefined {
    if (this.pendingReset || this.store.getStatus() === "suspended" || this.running.size > 0) return undefined;
    const [head] = this.heldTurns();
    if (!head) return undefined;
    const started = this.begin(head.turnId, { turnId: head.turnId, userText: head.userText, event: head.event, trigger: head.trigger });
    this.onDequeue?.(started.turnId, head);
    return started;
  }

  // A held turn is released when its turn settles — completed, failed, cancelled, or parked
  // (the park then owns it; resume continues it). A no-op for turns that were never held.
  private release(turnId: string): void {
    const held = this.heldTurns();
    if (!held.some((q) => q.turnId === turnId)) return;
    const rest = held.filter((q) => q.turnId !== turnId);
    this.store.tx(() => {
      this.store.delStep(INBOUND_QUEUE);
      if (rest.length) this.store.putStep(INBOUND_QUEUE, rest);
    });
  }

  // The turn this session is parked on, if any.
  private parkedTurnId(): string | undefined {
    if (this.store.getStatus() !== "suspended") return undefined;
    return (this.store.getStep("suspended") as SuspendedCheckpoint | undefined)?.turnId;
  }

  // Record an inbound turn durably (#263). Idempotent per turnId, like any inbound delivery:
  // a redelivery is not held again if the turn is already held OR has already started here (its
  // opening is in the log) — re-running a turn that already ran would render another turn's
  // tail as its reply. putStep is insert-only, so the list is replaced (delete + put) in a tx.
  private hold(turnId: string, input: TurnInput): void {
    const held = this.heldTurns();
    if (held.some((q) => q.turnId === turnId) || this.store.hasOpeningMessage(turnId)) return;
    const event = input.event ? { ...input.event, raw: undefined } : undefined;
    const entry: QueuedTurn = { turnId, userText: input.userText, event, trigger: input.trigger, hostContext: input.hostContext, queuedAt: new Date().toISOString() };
    const parked = this.store.getStatus() === "suspended" ? (this.store.getStep("suspended") as SuspendedCheckpoint | undefined) : undefined;
    this.store.tx(() => {
      this.store.delStep(INBOUND_QUEUE);
      this.store.putStep(INBOUND_QUEUE, [...held, entry]);
      if (parked && this.onAnnounce) {
        recordAnnouncement(this.store, announcement({ agent: this.agent, sessionId: this.id }, { kind: "held", turnId: parked.turnId, inputId: parked.request.id, queued: heldBehind(this.store, parked.turnId) }));
      }
    });
    void this.flushAnnouncements();
  }

  // Deliver the outbox to onAnnounce (#260), oldest first, removing each entry once the hook
  // returns. A hook that throws stops the flush and leaves the rest recorded for the next
  // one — triggered by the next announcement, the next settled turn, or a host rebuilding the
  // session. Never rejects: delivery problems are logged, not raised into the turn.
  // A call made while a flush is in flight joins it and asks for one more pass, which starts
  // only after the running one has fully finished (succeeded or failed) — so an announcement
  // recorded at any moment of a flush, even as it was ending, is never left behind, and the
  // returned promise settles only after that pass.
  flushAnnouncements(): Promise<void> {
    if (this.flushing) {
      this.flushAgain = true;
      return this.flushing;
    }
    // Nothing to deliver (or nobody to deliver to): stay settled. Every turn ends with a
    // flush, and a flush in flight keeps the actor from being idle() — an empty one must not.
    if (!this.onAnnounce || this.outbox().length === 0) return Promise.resolve();
    this.flushAgain = false;
    const pass = this.deliverOutbox()
      .catch((err) => { console.error(`[june] agent "${this.agent}" session "${this.id}": delivering an input announcement failed (kept for the next flush):`, err); })
      .then((): Promise<void> | void => {
        this.flushing = undefined;
        if (this.flushAgain) return this.flushAnnouncements();
      });
    this.flushing = pass;
    return pass;
  }
  private async deliverOutbox(): Promise<void> {
    for (;;) {
      const hook = this.onAnnounce;
      const [head] = this.outbox();
      if (!hook || !head) return;
      await hook(head);
      this.store.tx(() => {
        const rest = this.outbox().filter((a) => a.id !== head.id);
        this.store.delStep(ANNOUNCE_OUTBOX);
        if (rest.length) this.store.putStep(ANNOUNCE_OUTBOX, rest);
      });
    }
  }

  // Announcements recorded but not yet delivered, oldest first.
  undeliveredAnnouncements(): InputAnnouncement[] {
    return this.outbox();
  }
  private outbox(): InputAnnouncement[] {
    return (this.store.getStep(ANNOUNCE_OUTBOX) as InputAnnouncement[] | undefined) ?? [];
  }

  // Terminally retire this session's history (#129): supersede every unfinished turn,
  // then — ON THE CHAIN, so a turn started after this call lands in the fresh
  // generation by construction — archive messages/steps/status via store.reset() and
  // return an audit handle naming the archived generation. The session's ADDRESS never
  // changes (one DO is one session); what retires is its accumulated state: the next
  // turn starts from an empty transcript (no initiator, no suspended park — a stale
  // park is archived with the rest, which is the escape hatch for a dead approval).
  // The archived rows stay in the store's archive tables — that is the audit trail.
  reset(): Promise<{ previousSession: string; generation: number }> {
    const store = this.store;
    if (!store.reset) {
      return Promise.reject(new Error(`agent "${this.agent}" session "${this.id}": this SessionStore does not implement reset()`));
    }
    this.cancelPending("reset");
    // pendingReset opens the gate for turns queued BEHIND this archival (start() waives
    // its suspended check) and closes resume(); cleared when the op settles either way —
    // on failure the park still stands, and start()'s run-time guard protects it again.
    this.pendingReset = true;
    const op = this.chain.then(() => {
      // A reset retires the park (#260) — say so — and archives every step, the outbox among
      // them: carry what is undelivered into the new generation. The carry runs INSIDE the
      // reset's transaction (store.reset's callback, after the archive), so the archive and
      // the announcements commit together: a crash can lose neither without the other.
      const parked = store.getStatus() === "suspended" ? (store.getStep("suspended") as SuspendedCheckpoint | undefined) : undefined;
      const undelivered = this.outbox();
      const carry = this.onAnnounce && (parked || undelivered.length)
        ? () => {
            if (undelivered.length) store.putStep(ANNOUNCE_OUTBOX, undelivered);
            if (parked) recordAnnouncement(store, announcement({ agent: this.agent, sessionId: this.id }, { kind: "resolved", turnId: parked.turnId, inputId: parked.request.id, outcome: "retired" }));
          }
        : undefined;
      const generation = store.reset!(carry);
      if (parked) this.sink.emit({ type: "input.resolved", turnId: parked.turnId, inputId: parked.request.id, outcome: "retired" });
      void this.flushAnnouncements();
      return { previousSession: `${this.id}#g${generation}`, generation };
    });
    const clear = () => { this.pendingReset = false; };
    op.then(clear, clear);
    this.chain = op.catch(() => {}); // a failed reset must not break the inbox
    return op;
  }

  // Sugar: run a turn and await its final text. Explicitly the NON-INTERACTIVE convenience
  // (CLI, tests, simple cases) — it throws if the turn fails; use start()+observe()+result()
  // for liveness/interaction.
  // Not for held turns (#263): a held turn has no result to await yet — use start().
  turn(input: Omit<TurnInput, "ifSuspended" | "hostContext">): Promise<string> {
    const started = this.start(input);
    if (started.queued) return Promise.reject(new Error(`turn(): turn ${started.turnId} was held, not run — use start() for ifSuspended "queue"`));
    return this.running.get(started.turnId)!;
  }

  // Subscribe to this session's TurnEvent stream. Scope to one turn with `turnId`; with
  // `replay`, first emit the structural events foldable from the durable log (message.completed,
  // action.requested/completed, and turn.completed if finished) so a late/reconnecting subscriber
  // catches up, THEN live events. `turn.started`'s trigger + `turn.failed` are live-only (not
  // persisted), so they aren't part of the folded prefix.
  observe(cb: (e: TurnEvent) => void, opts?: { turnId?: string; replay?: boolean }): () => void {
    // Fold-then-subscribe is race-free HERE despite looking like a gap: this method is fully
    // SYNCHRONOUS (store.messages() + cb are sync) and the engine emits synchronously right
    // after each store.tx commit (no await between). Single-threaded JS therefore can't run an
    // engine step between the fold and the subscribe, so no event is missed or duplicated. (A
    // subscribe-first + buffer approach would instead double-deliver any event committed-before-
    // fold-but-emitted-after-subscribe.) If either invariant ever changes, revisit this.
    if (opts?.replay && opts.turnId) for (const e of this.foldEvents(opts.turnId)) cb(e);
    const { turnId } = opts ?? {};
    return this.sink.subscribe(turnId ? (e) => { if (e.turnId === turnId) cb(e); } : cb);
  }

  // Reconstruct the structural TurnEvents for one turn from the message log (the durable
  // counterpart of the live stream). Tool inputs are recovered from the assistant message's
  // tool calls so action.completed carries the full call.
  private foldEvents(turnId: string): TurnEvent[] {
    const msgs = this.store.messages().filter((m) => m.turnId === turnId);
    const inputs = new Map<string, ToolCall>();
    const out: TurnEvent[] = [];
    for (const m of msgs) {
      if (m.role === "assistant") {
        if (m.text.trim()) out.push({ type: "message.completed", turnId, text: m.text });
        for (const call of m.toolCalls) { inputs.set(call.id, call); out.push({ type: "action.requested", turnId, call }); }
      } else if (m.role === "tool") {
        out.push({
          type: "action.completed",
          turnId,
          call: inputs.get(m.toolCallId) ?? { id: m.toolCallId, name: m.name, input: undefined },
          result: m.result,
          ...(m.isError ? { error: String((m.result as { error?: unknown } | undefined)?.error ?? "tool failed") } : {}),
        });
        if (!m.isError) for (const file of fileRefsIn(m.result)) out.push({ type: "file.attached", turnId, callId: m.toolCallId, file });
      }
    }
    // terminal iff the last assistant message has no tool calls — same condition the live
    // engine uses (regardless of whether the final text is empty), so replay and live agree.
    const lastAssistant = [...msgs].reverse().find((m): m is Extract<Msg, { role: "assistant" }> => m.role === "assistant");
    if (lastAssistant && lastAssistant.toolCalls.length === 0)
      out.push({ type: "turn.completed", turnId, text: lastAssistant.text });
    return out;
  }

  transcript(): Turn[] { return foldTranscript(this.store.messages()); }
  snapshot() { return { transcript: this.transcript(), status: this.store.getStatus() }; }

  // True when no in-memory state of this actor matters any more: no turn running or
  // queued, no reset pending, no announcement flush in flight (#260 — the entry being
  // delivered is still in the outbox, so a rebuilt actor would hand it to the hook again,
  // in parallel). Everything else (transcript, steps, a suspended park) is in the store, so
  // a host may drop an idle actor and rebuild it on the next use (#174).
  // Live subscribers are held by the host's sink, not here — the host checks those.
  idle(): boolean {
    return this.running.size === 0 && !this.pendingReset && this.flushing === undefined;
  }
}

// Address space: a runtime resolves a session actor by (agentName, id). Native =
// an in-process map (@junejs/server); edge = idFromName on a DO namespace.
export interface Runtime {
  session(agent: string, id: string): AgentSession;
}

// Evaluate a parked input's { policy } answerer for a resumer, BEFORE the synchronous resume —
// the one async step a host takes (the authorize hook may query the db), kept apart so
// resume-then-subscribe stays free of awaits. Returns the grant to pass as resume's `granted`,
// or undefined: a { user } answerer needs no grant (resume checks `by`), and a refused or
// unevaluated policy yields none, so resume refuses with ResumeAuthorizationError. The grant
// names the policy it was made for; if the park changes in between, resume refuses it.
export async function grantAnswer(
  session: AgentSession,
  resume: { turnId: string; inputId: string; by?: string; principal?: Principal },
  authorize: AuthorizeAnswer | undefined,
): Promise<Answerers | undefined> {
  const pending = session.pending();
  if (!pending || pending.turnId !== resume.turnId || pending.request.id !== resume.inputId) return undefined;
  const answerers = pending.request.answerers;
  if (!answerers || "user" in answerers || !authorize) return undefined;
  const ok = await authorize({
    policy: answerers.policy,
    scope: answerers.scope,
    by: resume.by,
    principal: resume.principal,
    agent: session.agent,
    session: session.id,
    request: pending.request,
  });
  return ok === true ? answerers : undefined;
}