harn-stdlib 0.8.163

// @harn-entrypoint-category agent.stdlib
import { agent_autocompact_if_needed, agent_emergency_compact } from "std/agent/autocompact"
import { agent_budget_post_call_blocked, agent_budget_pre_call_blocked } from "std/agent/budget"
import {
  agent_loop_apply_command,
  agent_loop_control_invoke,
  agent_loop_snapshot_state,
} from "std/agent/control"
import { agent_daemon_snapshot, agent_daemon_step } from "std/agent/daemon"
import {
  agent_done_judge_cap,
  agent_verify_completion_judge_cap,
  agent_verify_or_continue,
} from "std/agent/judge"
import { agent_mcp_bootstrap_if_needed } from "std/agent/mcp"
import { agent_loop_options, agent_tool_format } from "std/agent/options"
import {
  agent_apply_tool_surface_narrowing,
  agent_compute_post_turn,
  agent_compute_terminal_callback,
  agent_reset_tool_surface_narrowing,
} from "std/agent/postturn"
import {
  agent_build_turn_messages,
  agent_build_turn_system_fragments,
  agent_tool_call_paradigm,
} from "std/agent/preflight"
import { agent_dispatch_tool_call } from "std/agent/primitives"
import { agent_progress_apply_options } from "std/agent/progress"
import { native_tool_contract_feedback_prompt, parse_guidance_prompt } from "std/agent/prompts"
import {
  agent_required_tools_enforce,
  agent_required_tools_inject_feedback,
  agent_required_tools_missing_from_session,
} from "std/agent/required_tools"
import { agent_scratchpad_init, agent_scratchpad_reorganize_if_due } from "std/agent/scratchpad"
import { agent_skills_match } from "std/agent/skills"
import { agent_sanitize_model_options } from "std/agent/stack"
import {
  agent_stall_apply_result,
  agent_stall_clear_current_failure,
  agent_stall_done_judge_due,
  agent_stall_initial_state,
  agent_stall_inject_feedback,
  agent_stall_no_net_progress,
  agent_stall_observe_tool_calls,
  agent_stall_repair_config,
} from "std/agent/stall"
import {
  agent_emit_event,
  agent_record_native_tool_fallback,
  agent_reminder_providers_fire,
  agent_session_drain_bridge_injections,
  agent_session_drain_feedback,
  agent_session_finalize,
  agent_session_init,
  agent_session_inject_feedback,
  agent_session_messages,
  agent_session_pop_last_assistant,
  agent_session_record_assistant,
  agent_session_record_tool_results,
  agent_session_record_usage,
  agent_session_totals,
} from "std/agent/state"
import { agent_step_judge } from "std/agent/step_judge"
import {
  agent_tool_search_emit_queries,
  agent_tool_search_inject_if_needed,
  agent_tool_search_record_results,
} from "std/agent/tool_search"
import {
  agent_await_resumption,
  agent_lifecycle_tools,
  list_agents,
  suspend_agent,
} from "std/agent/workers"
import { omit } from "std/json"
import { llm_call_options } from "std/llm/options"

/**
 * Pipeline entrypoints execute agent loops without a lexical `Harness` binding,
 * so loop budget timing uses the ambient builtins that share the VM clock backend.
 */
fn __agent_loop_clock_now_ms() {
  return now_ms()
}

fn __agent_loop_clock_sleep_ms(delay_ms) {
  sleep_ms(delay_ms)
}

fn __default_invoke_llm(message, turn_system, llm_opts) {
  let result = try {
    llm_call(message, turn_system, llm_call_options(llm_opts))
  }
  if !is_err(result) {
    return {ok: true, value: unwrap(result)}
  }
  let err = unwrap_err(result)
  let normalized = __llm_provider_error(err, llm_opts)
  let reason = if type_of(err) == "dict" {
    err?.reason ?? ""
  } else {
    ""
  }
  if reason == "budget_exceeded" {
    return {ok: false, status: "budget_exhausted", stop_reason: "budget_exhausted", error: normalized}
  }
  return {ok: false, status: "provider_error", stop_reason: "provider_error", error: normalized}
}

fn __messages_include_tool_result(messages) {
  if type_of(messages) != "list" {
    return false
  }
  for msg in messages {
    let role = to_string(msg?.role ?? "")
    if role == "tool" || role == "tool_result" {
      return true
    }
  }
  return false
}

fn __llm_error_field(err, key, fallback) {
  if type_of(err) == "dict" {
    let value = err[key]
    if value != nil {
      return to_string(value)
    }
  }
  return fallback
}

fn __llm_provider_error(err, llm_opts) {
  let message = __llm_error_field(err, "message", to_string(err))
  let category = __llm_error_field(err, "category", error_category(err))
  let reason = __llm_error_field(err, "reason", "")
  let kind = __llm_error_field(err, "kind", "")
  let provider = __llm_error_field(err, "provider", to_string(llm_opts?.provider ?? ""))
  let model = __llm_error_field(err, "model", to_string(llm_opts?.model ?? ""))
  return {
    category: category,
    reason: reason,
    kind: kind,
    provider: provider,
    model: model,
    message: message,
    phase: "llm_call",
    tool_format: to_string(llm_opts?.tool_format ?? ""),
    after_tool_result: __messages_include_tool_result(llm_opts?.messages),
  }
}

fn __agent_loop_emit_llm_call_start(llm_opts) {
  let session_id = to_string(llm_opts?.session_id ?? "")
  if session_id == "" {
    return nil
  }
  let checkpoint = {
    kind: "llm_call_start",
    phase: "llm_call",
    iteration: to_int(llm_opts?._iteration ?? 0),
    attempt: to_int(llm_opts?._truncation_auto_continue_attempt ?? 0) + 1,
    context_overflow_recovery_attempt: to_int(llm_opts?._context_overflow_recovery_attempt ?? 0),
    provider: to_string(llm_opts?.provider ?? ""),
    model: to_string(llm_opts?.model ?? ""),
    tool_format: to_string(llm_opts?.tool_format ?? ""),
    final_wrapup: llm_opts?._final_wrapup ?? false,
  }
  agent_emit_event(session_id, "typed_checkpoint", checkpoint)
  return nil
}

fn __validate_caller_result(r) {
  if type_of(r) != "dict" {
    throw "agent_loop: llm_caller must return a dict; got " + type_of(r)
  }
  if r?.ok == nil {
    throw "agent_loop: llm_caller result missing `ok`"
  }
  if r.ok && type_of(r?.value) != "dict" {
    throw "agent_loop: llm_caller returned ok=true but `value` is not a dict"
  }
  if !r.ok && type_of(r?.status) != "string" {
    throw "agent_loop: llm_caller returned ok=false but `status` is not a string"
  }
}

fn __invoke_llm(message, turn_system, llm_opts) {
  __agent_loop_emit_llm_call_start(llm_opts)
  let caller = llm_opts?._llm_caller
  if caller == nil {
    return __default_invoke_llm(message, turn_system, llm_opts)
  }
  let call = {
    prompt: message,
    system: turn_system,
    opts: llm_opts,
    turn: {iteration: llm_opts?._iteration ?? 0, session_id: llm_opts?.session_id ?? "", attempt: 1},
  }
  let result = try {
    caller(call)
  }
  if is_err(result) {
    throw unwrap_err(result)
  }
  let r = unwrap(result)
  __validate_caller_result(r)
  return r
}

// -------------------------------------------------------------------------------------------------

// Length-truncation auto-continue.
//
// When a value model hits the output-token cap mid-emit, the provider returns a
// length-truncation stop_reason (`length` for OpenAI/OpenRouter/Ollama,
// `max_tokens` for Anthropic) and the partial output often holds a TRUNCATED,
// unparseable tool call. Treating that as a malformed/missing call burns the
// turn on parse-guidance even though the model was mid-correct-action — a
// silent-corruption class that hurts even capable models.
//
// Instead we detect the specific condition deterministically (no model
// cooperation, no abuse surface) and re-issue the completion with a RAISED cap
// so the model can finish the call. The retry is invisible to the rest of the
// loop body: it does not consume an iteration or run stall/parse accounting.
// Bounded to a small number of continuations; if still truncated after the cap
// we return the last result and the existing parse-guidance path takes over.
//
// The tool-call gate (`__host_agent_truncated_tool_call`) fires ONLY on a real
// length truncation with zero usable calls AND a partial-call signal. A second
// gate covers hidden-reasoning-only truncation: strict providers can return no
// visible text/tool calls, `stop_reason: length`, and non-empty reasoning. That
// is still a budget exhaustion, so retry with a larger output cap instead of
// handing an empty turn to the normal loop body.

// -------------------------------------------------------------------------------------------------

fn __autocontinue_max_continuations(llm_opts) {
  let configured = llm_opts?.truncation_auto_continue_max
  if type_of(configured) == "int" && configured >= 0 {
    return configured
  }
  return 2
}

/**
 * Compute the raised output-token cap for an auto-continue retry. An unset cap
 * (`<= 0`) means the provider's own default truncated us, so jump to an
 * explicit base. A set cap doubles. Both are clamped to a ceiling so a
 * misconfigured model can't request an unbounded body.
 */
fn __autocontinue_raised_cap(current, llm_opts) {
  let base = if type_of(llm_opts?.truncation_auto_continue_base) == "int"
    && llm_opts.truncation_auto_continue_base > 0 {
    llm_opts.truncation_auto_continue_base
  } else {
    16384
  }
  let ceiling = if type_of(llm_opts?.truncation_auto_continue_ceiling) == "int"
    && llm_opts.truncation_auto_continue_ceiling > 0 {
    llm_opts.truncation_auto_continue_ceiling
  } else {
    32768
  }
  let raised = if type_of(current) == "int" && current > 0 {
    current * 2
  } else {
    base
  }
  if raised > ceiling {
    return ceiling
  }
  return raised
}

fn __autocontinue_is_length_stop(stop_reason) {
  let normalized = lowercase(trim(to_string(stop_reason ?? "")))
  return normalized == "length" || normalized == "max_tokens"
}

fn __autocontinue_hidden_reasoning_truncated(llm_result, tool_call_count) {
  if !__autocontinue_is_length_stop(llm_result?.stop_reason) {
    return false
  }
  if tool_call_count > 0 {
    return false
  }
  if trim(to_string(llm_result?.text ?? "")) != "" {
    return false
  }
  let thinking = trim(to_string(llm_result?.thinking ?? ""))
  let summary = trim(to_string(llm_result?.thinking_summary ?? ""))
  if thinking != "" || summary != "" {
    return true
  }
  return (to_int(llm_result?.output_tokens ?? 0) ?? 0) > 0
}

/**
 * Detect whether `call` is a length-truncated turn that resolved no usable
 * tool call but looks like it was mid-call — the one condition where
 * re-issuing with a raised cap is the right move.
 */
fn __autocontinue_should_continue(call, turn_opts) {
  if !(call?.ok ?? false) {
    return false
  }
  let llm_result = call?.value
  if type_of(llm_result) != "dict" {
    return false
  }
  let raw_text = llm_result?.text ?? ""
  let parsed = agent_parse_tool_calls(raw_text, turn_opts?.tools, turn_opts?.tool_format)
  let tool_calls = __resolve_tool_calls(llm_result, parsed)
  let has_parse_errors = len(parsed?.tool_parse_errors ?? []) > 0
  if __host_agent_truncated_tool_call(
    llm_result?.stop_reason,
    raw_text,
    len(tool_calls),
    has_parse_errors,
  ) {
    return true
  }
  return __autocontinue_hidden_reasoning_truncated(llm_result, len(tool_calls))
}

/**
 * Call the LLM, and on a length-truncated turn with an incomplete tool call,
 * AUTO-CONTINUE by re-issuing with a raised output cap (bounded). Returns the
 * same `{ok, value, ...}` shape `__invoke_llm` returns; callers consume the
 * final (hopefully complete) result and run their normal parse/stall/dispatch
 * accounting on it. Falls back to the last truncated result once the cap is
 * exhausted, so the existing parse-guidance path still fires.
 */
fn __invoke_llm_with_autocontinue(
  message,
  turn_system,
  llm_opts,
  turn_opts,
  session_id,
  iteration_index,
) {
  var call = __invoke_llm(message, turn_system, llm_opts)
  let max_continuations = __autocontinue_max_continuations(llm_opts)
  var attempts = 0
  var cap = llm_opts?.max_tokens ?? 0
  while attempts < max_continuations && __autocontinue_should_continue(call, turn_opts) {
    let raised = __autocontinue_raised_cap(cap, llm_opts)
    // No headroom left to raise — re-issuing would just truncate again at the
    // same cap, so stop and let parse-guidance take the turn.
    if type_of(cap) == "int" && cap > 0 && raised <= cap {
      break
    }
    attempts = attempts + 1
    agent_emit_event(
      session_id,
      "llm_auto_continue",
      {
        iteration: iteration_index + 1,
        attempt: attempts,
        max_continuations: max_continuations,
        previous_max_tokens: cap,
        raised_max_tokens: raised,
        stop_reason: call?.value?.stop_reason ?? "",
      },
    )
    cap = raised
    let retry_opts = llm_opts + {max_tokens: raised, _truncation_auto_continue_attempt: attempts}
    call = __invoke_llm(message, turn_system, retry_opts)
  }
  return call
}

// -------------------------------------------------------------------------------------------------

/**
 * Context-overflow recovery.
 *
 * A provider can reject a turn with a `context_overflow` error when the
 * assembled prompt exceeds the model's real context window — typically on a
 * large repo where tool observations accreted past the budget, OR when the
 * model's window is mis/under-cataloged so auto-compaction never fired. The
 * agent must NOT die on this: it is a recoverable, self-inflicted condition.
 *
 * Recovery is a bounded loop: emergency-compact the live transcript
 * (deterministic observation masking — never an LLM call, which would itself
 * overflow), then re-issue the SAME turn. Each attempt compacts more
 * aggressively (shorter preserved tail). We stop when either the retry
 * succeeds, or emergency compaction can no longer shrink the transcript
 * (`archived == 0`, i.e. an irreducible system-prompt + single oversized
 * message), at which point the overflow is genuinely terminal.
 */
fn __agent_loop_is_context_overflow(error) {
  if type_of(error) != "dict" {
    return false
  }
  let reason = to_string(error?.reason ?? "")
  if reason == "context_overflow" {
    return true
  }
  // Defensive fallback: some routes surface the condition only in the message
  // text (uncataloged provider, no structured `reason`). Match the canonical
  // classifier tag the Rust LLM layer stamps onto the error string.
  let message = to_string(error?.message ?? "")
  return contains(message, "[context_overflow]")
}

fn __agent_loop_context_overflow_max_recoveries(llm_opts) {
  let configured = llm_opts?.context_overflow_recover_max
  if type_of(configured) == "int" && configured >= 0 {
    return configured
  }
  return 3
}

/**
 * On a `context_overflow` provider error, emergency-compact + retry (bounded).
 * Returns a fresh `{ok, ...}` call result: `ok:true` when a retry succeeded,
 * or the last `ok:false` result (so the caller's terminal path runs unchanged)
 * when recovery is exhausted or the transcript is irreducible.
 */
fn __agent_loop_recover_context_overflow(
  call,
  message,
  turn_system,
  llm_opts,
  turn_opts,
  session,
  iteration_index,
) {
  let max_recoveries = __agent_loop_context_overflow_max_recoveries(llm_opts)
  var current = call
  var attempt = 0
  while attempt < max_recoveries && !current.ok
    && __agent_loop_is_context_overflow(current?.error) {
    attempt = attempt + 1
    let archived = agent_emergency_compact(session, llm_opts, attempt)
    agent_emit_event(
      session.session_id,
      "context_overflow_recovery",
      {
        iteration: iteration_index + 1,
        attempt: attempt,
        max_recoveries: max_recoveries,
        archived_messages: archived,
        provider_error: current?.error ?? {},
      },
    )
    // Nothing left to shed: the prompt is irreducible (system prompt + a single
    // oversized message). Re-issuing would overflow identically, so surface the
    // original terminal error rather than spin.
    if archived <= 0 {
      break
    }
    // Rebuild the turn prompt against the now-compacted transcript and re-issue.
    let turn_prompt = __agent_loop_build_turn_prompt(session, llm_opts, iteration_index)
    let retry_opts = llm_opts
      + {
      messages: turn_prompt.messages,
      _system_fragments: turn_prompt.fragments,
      _context_overflow_recovery_attempt: attempt,
    }
    current = __invoke_llm_with_autocontinue(
      message,
      turn_prompt.system,
      retry_opts,
      turn_opts,
      session.session_id,
      iteration_index,
    )
  }
  return current
}

// -------------------------------------------------------------------------------------------------

// Tool middleware seam — composable tool_caller (mirrors __invoke_llm).
//
// Each tool dispatch is funneled through `tool_caller(envelope, next)` when
// the agent_loop options carry one. The envelope normalizes the call shape
// so middleware doesn't have to peek at the underlying registry/schema:
//
//   envelope = {
//     tool_name, tool_args, call_id,
//     declared_executor?, schema?, description?,
//     turn: {iteration, session_id},
//   }
//
// The middleware returns a dispatch-shape dict. Calling `next(envelope)`
// runs the default dispatch (with any envelope mutations the middleware
// applied — typically `tool_args` rewrites or argument stripping). Callers
// can short-circuit by returning their own dict without invoking `next`.
//
// See std/llm/tool_middleware for the userspace primitives + the bundled
// middleware library (with_required_reason, with_audit_log, …).

// -------------------------------------------------------------------------------------------------

fn __tool_registry_entry(tools, tool_name) {
  if tools == nil {
    return nil
  }
  let entries = tools?.tools
  if type_of(entries) != "list" {
    return nil
  }
  for entry in entries {
    if type_of(entry) != "dict" {
      continue
    }
    let entry_name = if entry?.name != nil {
      to_string(entry.name)
    } else {
      let func = entry?.function
      if type_of(func) == "dict" {
        to_string(func?.name ?? "")
      } else {
        ""
      }
    }
    if entry_name == tool_name {
      return entry
    }
  }
  return nil
}

fn __tool_entry_annotations(entry) {
  if type_of(entry) != "dict" {
    return {}
  }
  var annotations = {}
  let policy = entry?.policy
  if type_of(policy) == "dict" {
    annotations = annotations + policy
  }
  let direct = entry?.annotations
  if type_of(direct) == "dict" {
    annotations = annotations + direct
  }
  let func = entry?.function
  if type_of(func) == "dict" && type_of(func?.policy) == "dict" {
    annotations = annotations + func.policy
  }
  if type_of(func) == "dict" && type_of(func?.annotations) == "dict" {
    annotations = annotations + func.annotations
  }
  return annotations
}

fn __tool_policy_annotations(policy, tool_name) {
  if type_of(policy) != "dict" {
    return {}
  }
  let registry = policy?.tool_annotations ?? policy?.toolAnnotations ?? {}
  if type_of(registry) != "dict" {
    return {}
  }
  let annotations = registry[tool_name]
  if type_of(annotations) == "dict" {
    return annotations
  }
  return {}
}

fn __tool_resource_annotations(entry, policy, tool_name) {
  return __tool_policy_annotations(policy, tool_name) + __tool_entry_annotations(entry)
}

fn __tool_envelope(call, tools, options) {
  let tool_name = to_string(call?.name ?? call?.tool_name ?? "")
  let tool_args_raw = call?.arguments ?? call?.tool_args
  let tool_args = if type_of(tool_args_raw) == "dict" {
    tool_args_raw
  } else {
    {}
  }
  let raw_call_id = to_string(call?.id ?? call?.tool_call_id ?? "")
  let call_id = if raw_call_id == "" {
    "tool_call_" + uuid()
  } else {
    raw_call_id
  }
  let entry = __tool_registry_entry(tools, tool_name)
  let declared_executor = if entry == nil {
    nil
  } else {
    let direct = entry?.executor
    if direct != nil {
      to_string(direct)
    } else {
      let func = entry?.function
      if type_of(func) == "dict" && func?.executor != nil {
        to_string(func.executor)
      } else {
        nil
      }
    }
  }
  let schema = if entry == nil {
    nil
  } else {
    entry?.parameters ?? entry?.input_schema ?? entry?.inputSchema
  }
  let annotations = __tool_resource_annotations(entry, options?.policy, tool_name)
  let description = if entry == nil {
    ""
  } else {
    let direct = entry?.description
    if direct != nil {
      to_string(direct)
    } else {
      let func = entry?.function
      if type_of(func) == "dict" && func?.description != nil {
        to_string(func.description)
      } else {
        ""
      }
    }
  }
  return {
    tool_name: tool_name,
    tool_args: tool_args,
    call_id: call_id,
    declared_executor: declared_executor,
    schema: schema,
    annotations: annotations,
    description: description,
    turn: {
      iteration: options?._iteration ?? 0,
      session_id: to_string(options?.session_id ?? ""),
      run_id: options?.run_id ?? options?._run_id,
      model: options?.model,
      provider: options?.provider,
      tool_call_index: options?._tool_call_index ?? 0,
      max_concurrent_tools: options?._max_concurrent_tools ?? 1,
      prefetch_next_turn: options?._prefetch_next_turn ?? false,
    },
  }
}

fn __default_invoke_tool(envelope, original_call, tools, options) {
  let next_call = original_call
    + {
    id: envelope.call_id,
    tool_call_id: envelope.call_id,
    name: envelope.tool_name,
    tool_name: envelope.tool_name,
    arguments: envelope.tool_args,
  }
  return agent_dispatch_tool_call(next_call, tools, options)
}

fn __validate_tool_caller_result(r) {
  if type_of(r) != "dict" {
    throw "agent_loop: tool_caller must return a dict; got " + type_of(r)
  }
  let name = r?.tool_name ?? r?.name
  if name == nil || to_string(name) == "" {
    throw "agent_loop: tool_caller result missing `tool_name`"
  }
  let ok = r?.ok
  if ok == nil {
    let success = r?.success
    if success == nil {
      let status = r?.status
      if status == nil {
        throw "agent_loop: tool_caller result missing `ok`/`success`/`status`"
      }
    }
  } else if type_of(ok) != "bool" {
    throw "agent_loop: tool_caller result `ok` must be a bool; got " + type_of(ok)
  }
}

fn __middleware_exception_result(envelope, err) {
  let err_text = to_string(err)
  let observation = "[error from " + envelope.tool_name + "]\n" + err_text
    + "\n[end of "
    + envelope.tool_name
    + " error]\n"
  return {
    ok: false,
    status: "error",
    tool_name: envelope.tool_name,
    tool_call_id: envelope.call_id,
    arguments: envelope.tool_args,
    result: nil,
    rendered_result: err_text,
    observation: observation,
    error: err_text,
    error_category: "tool_middleware_exception",
    executor: nil,
  }
}

fn __structural_validator_tool_name() -> string {
  return "__structural_validator_turn__"
}

fn __structural_validator_pass_result(envelope) {
  return {
    ok: true,
    status: "ok",
    tool_name: envelope.tool_name,
    tool_call_id: envelope.call_id,
    arguments: envelope.tool_args,
    result: {configured: false, vetoed: false, skipped: true, reason: "not_configured"},
    rendered_result: "",
    observation: "",
    error: nil,
    error_category: nil,
    executor: "harn",
  }
}

fn __run_structural_validator(
  caller,
  session_id,
  llm_result,
  tool_calls,
  parsed,
  llm_opts,
  turn_opts,
  prior_successful_tools,
  prior_rejected_tools,
  attempts,
) {
  if caller == nil {
    return {configured: false, vetoed: false, skipped: true, reason: "not_configured"}
  }
  let envelope = {
    tool_name: __structural_validator_tool_name(),
    tool_args: {
      session_id: session_id,
      iteration: turn_opts?._iteration ?? 0,
      attempts: attempts,
      tool_calls: tool_calls,
      tools: turn_opts?.tools,
      policy: turn_opts?.policy,
      assistant_text: llm_result?.visible_text ?? llm_result?.text ?? "",
      raw_text: llm_result?.raw_text ?? llm_result?.text ?? "",
      parsed_done_marker: parsed?.done_marker ?? "",
      tool_parse_errors: parsed?.tool_parse_errors ?? [],
      protocol_violations: parsed?.protocol_violations ?? [],
      tool_format: turn_opts?.tool_format ?? llm_opts?.tool_format ?? "",
      output_tokens: llm_result?.output_tokens ?? 0,
      max_output_tokens: llm_opts?.max_tokens ?? turn_opts?.max_tokens ?? 0,
      provider: llm_result?.provider ?? "",
      model: llm_result?.model ?? "",
      prior_successful_tools: prior_successful_tools,
      prior_rejected_tools: prior_rejected_tools,
    },
    call_id: "structural-validator-turn-" + to_string(turn_opts?._iteration ?? 0),
    declared_executor: "harn",
    schema: nil,
    annotations: nil,
    description: "Internal structural validator probe",
    turn: {
      iteration: turn_opts?._iteration ?? 0,
      session_id: session_id,
      run_id: turn_opts?.run_id ?? turn_opts?._run_id,
      model: turn_opts?.model,
      provider: turn_opts?.provider,
      tool_call_index: 0,
      max_concurrent_tools: 1,
      prefetch_next_turn: false,
    },
  }
  let next = { env_in -> __structural_validator_pass_result(env_in) }
  let outcome = try {
    caller(envelope, next)
  }
  if is_err(outcome) {
    let err = unwrap_err(outcome)
    if error_category(err) == "cancelled" {
      throw err
    }
    throw "agent_loop: structural validator failed: " + to_string(err)
  }
  let result = unwrap(outcome)
  if type_of(result) != "dict" {
    throw "agent_loop: structural validator must return a dict; got " + type_of(result)
  }
  return if type_of(result?.result) == "dict" {
    result.result
  } else {
    {}
  }
}

fn __tool_lifecycle_session_id(envelope) -> string {
  return to_string(envelope?.turn?.session_id ?? "")
}

fn __emit_tool_lifecycle_start(envelope) {
  let session_id = __tool_lifecycle_session_id(envelope)
  if session_id == "" || to_string(envelope?.tool_name ?? "") == "" {
    return
  }
  let _ = agent_emit_event(
    session_id,
    "tool_call",
    {
      tool_call_id: envelope.call_id,
      tool_name: envelope.tool_name,
      status: "pending",
      raw_input: envelope.tool_args,
    },
  )
  let _ = agent_emit_event(
    session_id,
    "tool_call_update",
    {
      tool_call_id: envelope.call_id,
      tool_name: envelope.tool_name,
      status: "in_progress",
      raw_input: envelope.tool_args,
    },
  )
}

fn __tool_terminal_status(result) -> string {
  if __tool_result_product_error(result) {
    return "failed"
  }
  if result?.ok || result?.success {
    return "completed"
  }
  let status = to_string(result?.status ?? "")
  if status == "ok" || status == "success" {
    return "completed"
  }
  return "failed"
}

fn __tool_lifecycle_error_category(raw) {
  if raw == nil {
    return nil
  }
  let category = to_string(raw)
  if category == "" {
    return nil
  }
  if contains(
    [
      "schema_validation",
      "tool_error",
      "mcp_server_error",
      "host_bridge_error",
      "permission_denied",
      "rejected_loop",
      "parse_aborted",
      "timeout",
      "network",
      "cancelled",
      "unknown",
    ],
    category,
  ) {
    return category
  }
  if contains(["intra_turn_failure_fanout_collapsed", "intra_turn_resource_fail_fast"], category) {
    return "rejected_loop"
  }
  if contains(["tool_rejected", "consent_denied", "egress_blocked", "scope_violation"], category) {
    return "permission_denied"
  }
  if contains(
    ["schema_violation", "schema_stream_aborted", "validator_failure", "repair_failed"],
    category,
  ) {
    return "schema_validation"
  }
  if contains(["rate_limit", "overloaded", "server_error", "transient_network"], category) {
    return "network"
  }
  if contains(["tool_middleware_exception", "tool_parallel_dispatch_exception", "generic"], category) {
    return "tool_error"
  }
  if contains(["auth", "channel_closed", "not_found", "circuit_open", "budget_exceeded"], category) {
    return "host_bridge_error"
  }
  return "unknown"
}

fn __emit_tool_lifecycle_finish(envelope, result) {
  let session_id = __tool_lifecycle_session_id(envelope)
  if session_id == "" || to_string(envelope?.tool_name ?? "") == "" {
    return
  }
  let result_tool_call_id = to_string(result?.tool_call_id ?? "")
  let tool_call_id = if result_tool_call_id != "" {
    result_tool_call_id
  } else {
    envelope.call_id
  }
  let tool_name = to_string(result?.tool_name ?? result?.name ?? envelope.tool_name)
  let raw_output = if result?.result != nil {
    result.result
  } else {
    result?.rendered_result ?? result?.output ?? result
  }
  let payload = {
    tool_call_id: tool_call_id,
    tool_name: tool_name,
    status: __tool_terminal_status(result),
    raw_output: raw_output,
    error: result?.error,
    duration_ms: result?.duration_ms ?? result?.execution_duration_ms,
    execution_duration_ms: result?.execution_duration_ms,
    error_category: __tool_lifecycle_error_category(result?.error_category),
    executor: result?.executor,
  }
  let _ = agent_emit_event(session_id, "tool_call_update", payload)
}

fn __emit_synthetic_tool_lifecycle_finish(call, result, tools, options) {
  let envelope = __tool_envelope(call, tools, options)
  __emit_tool_lifecycle_finish(envelope, result)
}

fn __invoke_tool(call, tools, options) {
  let caller = options?._tool_caller
  let envelope = __tool_envelope(call, tools, options)
  __emit_tool_lifecycle_start(envelope)
  if caller == nil {
    let direct = agent_dispatch_tool_call(
      call + {id: envelope.call_id, tool_call_id: envelope.call_id},
      tools,
      options,
    )
    __emit_tool_lifecycle_finish(envelope, direct)
    return direct
  }
  let next = { env_in -> __default_invoke_tool(env_in, call, tools, options) }
  let outcome = try {
    caller(envelope, next)
  }
  if is_err(outcome) {
    let err = unwrap_err(outcome)
    if error_category(err) == "cancelled" {
      throw err
    }
    __maybe_emit_tool_audit(
      envelope.turn.session_id,
      envelope,
      {layer: "tool_caller", status: "exception", error: to_string(err)},
    )
    let result = __middleware_exception_result(envelope, err)
    __emit_tool_lifecycle_finish(envelope, result)
    return result
  }
  let r = unwrap(outcome)
  __validate_tool_caller_result(r)
  __maybe_emit_tool_audit(envelope.turn.session_id, envelope, r?.audit, r?.receipt)
  __emit_tool_lifecycle_finish(envelope, r)
  return r
}

fn __maybe_emit_tool_audit(session_id, envelope, audit, receipt = nil) {
  if audit == nil && receipt == nil {
    return
  }
  if session_id == "" {
    return
  }
  let payload = if receipt == nil {
    {tool_call_id: envelope.call_id, tool_name: envelope.tool_name, audit: audit}
  } else {
    {
      tool_call_id: envelope.call_id,
      tool_name: envelope.tool_name,
      audit: audit ?? {},
      receipt: receipt,
    }
  }
  let _ = try {
    agent_emit_event(session_id, "tool_call_audit", payload)
  }
}

fn __visible_text(parsed, raw_text) {
  if parsed?.user_response != nil && parsed.user_response != "" {
    return parsed.user_response
  }
  if parsed?.prose != nil && parsed.prose != "" {
    return parsed.prose
  }
  return raw_text
}

/**
 * A tool call with an empty / whitespace-only name is provider JSON
 * malformation, not a dispatchable call. Some providers emit a stray
 * `{name: "", arguments: {}}` block alongside valid sibling calls (observed
 * live on go/rust eval runs: a turn carrying `[look, "", look]`). Dispatching
 * the nameless call resolves no tool and can otherwise terminate the agent
 * loop silently — a pure harness failure that reads as a model give-up
 * (result=INCOMPLETE, outcome_kind=null). Detect it here so callers can drop
 * the malformed call, keep the valid siblings, and inject parse-guidance.
 */
fn __tool_call_name_is_blank(call) -> bool {
  return __tool_call_name(call).trim() == ""
}

/**
 * Partition tool calls into the dispatchable ones (named) and a count of the
 * malformed empty-name calls that were dropped. Source order of the kept calls
 * is preserved so order-sensitive parsers/parallel dispatch stay correct.
 */
fn __filter_blank_name_tool_calls(calls) {
  var kept = []
  var dropped = 0
  for call in calls ?? [] {
    if __tool_call_name_is_blank(call) {
      dropped = dropped + 1
    } else {
      kept = kept.push(call)
    }
  }
  return {calls: kept, dropped: dropped}
}

fn __resolve_tool_calls(llm_result, parsed) {
  let native_calls = llm_result?.native_tool_calls ?? llm_result?.tool_calls ?? []
  if len(native_calls) > 0 {
    return __filter_blank_name_tool_calls(native_calls).calls
  }
  return __filter_blank_name_tool_calls(parsed?.calls ?? []).calls
}

fn __agent_await_resumption_call(tool_calls) {
  for call in tool_calls {
    if __tool_call_name(call) == "agent_await_resumption" {
      return call
    }
  }
  return nil
}

fn __agent_await_resumption_args(call) {
  let args = __tool_call_args(call)
  return agent_await_resumption(args?.reason ?? "", args?.conditions ?? nil, args?.resume_by ?? nil)
}

fn __agent_loop_await_resumption(session, iteration, call, opts) {
  let parsed = __agent_await_resumption_args(call)
  let worker = __agent_loop_current_worker()
  if worker == nil {
    return __agent_loop_await_resumption_top_level(session, iteration, call, parsed, opts)
  }
  agent_emit_event(
    session.session_id,
    "tool_call_audit",
    {
      tool_call_id: to_string(call?.id ?? call?.tool_call_id ?? ""),
      tool_name: "agent_await_resumption",
      audit: {
        layer: "agent_lifecycle",
        status: "suspended",
        initiator: "self",
        reason: parsed.reason,
        worker_id: worker?.id,
        conditions: parsed.conditions,
      },
    },
  )
  suspend_agent(worker, parsed.reason, {initiator: "self", conditions: parsed.conditions})
  let checkpoint = __agent_loop_suspend_checkpoint(session, iteration)
  if checkpoint == nil {
    throw "agent_await_resumption: suspend checkpoint did not yield"
  }
  return checkpoint
}

fn __agent_loop_await_resumption_top_level(session, iteration, call, parsed, opts) {
  agent_session_inject(
    session.session_id,
    transcript_reminder_event(
      {
        body: __agent_loop_suspend_reminder_body(parsed.reason),
        source: "in_pipeline",
        tags: ["agent_loop", "top_level_suspend"],
        dedupe_key: "top_level_suspend:" + session.session_id,
        ttl_turns: 1,
        fired_at_turn: iteration + 1,
      },
    ),
  )
  let handle = __host_top_level_agent_suspend(
    session.session_id,
    session.task,
    session.system,
    opts,
    parsed.reason,
    parsed.conditions,
    iteration,
  )
  agent_emit_event(
    session.session_id,
    "tool_call_audit",
    {
      tool_call_id: to_string(call?.id ?? call?.tool_call_id ?? ""),
      tool_name: "agent_await_resumption",
      audit: {
        layer: "agent_lifecycle",
        status: "suspended",
        initiator: "self",
        reason: parsed.reason,
        worker_id: handle?.id,
        conditions: parsed.conditions,
      },
    },
  )
  return {
    status: "suspended",
    handle: handle,
    worker: handle,
    reason: parsed.reason,
    initiator: "self",
    conditions: parsed.conditions,
    resume_by: parsed?.resume_by,
    iterations_completed: iteration,
    session_id: session.session_id,
  }
}

fn __agent_loop_invalid_await_resumption_feedback(error) {
  return "Invalid agent_await_resumption call: "
    + to_string(error)
    + "\n\nThis lifecycle tool is only for parking until external input or a valid resume condition. "
    + "For ordinary work, continue with the available project tools or finish normally. "
    + "If you do need a resume condition, `conditions.trigger` must be an object trigger spec; "
    + "use `conditions.on_event` for an event-topic string."
}

fn __tool_call_name(call) {
  return to_string(call?.name ?? call?.tool_name ?? "")
}

fn __tool_call_args(call) {
  let raw = call?.arguments ?? call?.tool_args
  if type_of(raw) == "dict" {
    return raw
  }
  return {}
}

fn __native_fallback_feedback(policy, fallback_index) {
  return native_tool_contract_feedback_prompt({policy: policy, fallback_index: fallback_index})
}

fn __parse_feedback_tool_annotations(entry) {
  return __tool_entry_annotations(entry)
}

fn __parse_feedback_annotation_enabled(value) -> bool {
  return type_of(value) == "bool" && value
}

fn __parse_feedback_tool_is_structural(entry) -> bool {
  let annotations = __parse_feedback_tool_annotations(entry)
  return __parse_feedback_annotation_enabled(annotations?.structural)
    || __parse_feedback_annotation_enabled(annotations?.agent_lifecycle)
}

fn __parse_feedback_has_non_structural_tools(turn_opts) -> bool {
  let entries = turn_opts?.tools?.tools ?? []
  for entry in entries {
    if type_of(entry) == "dict" && !__parse_feedback_tool_is_structural(entry) {
      return true
    }
  }
  return false
}

/**
 * A turn whose tool calls were ALL dropped by the parser, or whose response
 * violates the tagged protocol with 0 dispatched calls, is a malformed-output
 * turn, not a no-progress monologue. The
 * purpose-built `parse_guidance` prompt — which shows the heredoc syntax and
 * names the exact parser diagnostic — is the right corrective feedback, but
 * nothing consumed `tool_parse_errors` on the active path. This fires purely on
 * the syntactic parse-error condition: strong models emit clean calls, hit zero
 * parse errors, and never reach it (no regression).
 *
 * Partial-success turns (some calls parsed AND at least one was dropped) get
 * the same parse_guidance note — flagged `has_partial_success` so the model
 * knows the good calls dispatched and only re-emits the malformed one. Without
 * this, the dropped call's diagnostic was silently swallowed and the model got
 * zero signal to re-emit it.
 *
 * Returns true ONLY when the WHOLE turn was a parse failure (zero calls
 * dispatched) so the caller can suppress the no-progress stall path. A
 * partial-success turn made real progress (its parsed calls dispatched), so it
 * returns false and stays subject to the normal stall accounting.
 */
fn __parse_feedback_protocol_violations(parsed, turn_opts) {
  if agent_tool_call_paradigm(turn_opts).kind != "text"
    || !__parse_feedback_has_non_structural_tools(turn_opts) {
    return []
  }
  return parsed?.protocol_violations ?? []
}

fn __parse_feedback_diagnostics(parsed, turn_opts) {
  let parse_errors = parsed?.tool_parse_errors ?? []
  let protocol_violations = __parse_feedback_protocol_violations(parsed, turn_opts)
  return parse_errors + protocol_violations
}

fn __maybe_inject_parse_error_feedback(session_id, parsed, tool_calls, turn_opts) -> bool {
  let protocol_violations = __parse_feedback_protocol_violations(parsed, turn_opts)
  let diagnostics = __parse_feedback_diagnostics(parsed, turn_opts)
  if len(diagnostics) == 0 {
    return false
  }
  // Some calls parsed AND at least one was dropped: the good calls already
  // dispatched, so flag the note as partial success and report how many landed
  // so the model only re-emits the malformed call.
  let parsed_count = len(tool_calls)
  let has_partial_success = parsed_count > 0
  let error_summary = to_string(diagnostics[0])
  let feedback = parse_guidance_prompt(
    {
      error_summary: error_summary,
      has_partial_success: has_partial_success,
      parsed_call_count: parsed_count,
      body_hint: agent_tool_call_paradigm(turn_opts).body_hint,
      is_native_format: agent_tool_format(turn_opts) == "native",
      is_json_format: agent_tool_format(turn_opts) == "json",
    },
    turn_opts,
  )
  agent_session_inject_feedback(session_id, "parse_guidance", feedback)
  agent_emit_event(
    session_id,
    "tool_parse_error_feedback",
    {
      parse_error_count: len(parsed?.tool_parse_errors ?? []),
      protocol_violation_count: len(protocol_violations),
      diagnostic_count: len(diagnostics),
      error_summary: error_summary,
      has_partial_success: has_partial_success,
      parsed_call_count: parsed_count,
    },
  )
  // Only a full parse drop (zero calls dispatched) suppresses the no-progress
  // stall path; a partial-success turn made real progress and stays subject to
  // normal stall accounting.
  return !has_partial_success
}

/**
 * Count empty-name tool calls in the raw turn (native, else text-parsed) that
 * `__resolve_tool_calls` will drop. Used by the dispatch site to inject
 * parse-guidance so the model re-emits a valid call next turn instead of the
 * loop terminating on the malformed sibling.
 */
fn __blank_name_dropped_count(llm_result, parsed) -> int {
  let native_calls = llm_result?.native_tool_calls ?? llm_result?.tool_calls ?? []
  let source = if len(native_calls) > 0 {
    native_calls
  } else {
    parsed?.calls ?? []
  }
  return __filter_blank_name_tool_calls(source).dropped
}

/**
 * Drop-only guidance for provider-malformed empty-name tool calls. The
 * filtered valid siblings have already dispatched; this tells the model the
 * nameless call was discarded and to re-emit a named call. Returns true when a
 * note was injected (so the caller flags turn-level tool-call feedback and the
 * stall accounting stays consistent with the parse-error path).
 */
fn __maybe_inject_blank_name_feedback(session_id, llm_result, parsed, dispatched_count, turn_opts) -> bool {
  let dropped = __blank_name_dropped_count(llm_result, parsed)
  if dropped == 0 {
    return false
  }
  let has_partial_success = dispatched_count > 0
  let feedback = parse_guidance_prompt(
    {
      error_summary: "Dropped "
        + to_string(dropped)
        + " tool call(s) with an empty/blank name — every tool call must name a tool.",
      has_partial_success: has_partial_success,
      parsed_call_count: dispatched_count,
      body_hint: agent_tool_call_paradigm(turn_opts).body_hint,
      is_native_format: agent_tool_format(turn_opts) == "native",
      is_json_format: agent_tool_format(turn_opts) == "json",
    },
    turn_opts,
  )
  agent_session_inject_feedback(session_id, "parse_guidance", feedback)
  agent_emit_event(
    session_id,
    "tool_call_blank_name_dropped",
    {
      dropped_count: dropped,
      dispatched_count: dispatched_count,
      has_partial_success: has_partial_success,
    },
  )
  return true
}

fn __detect_native_fallback(
  llm_result,
  parsed,
  turn_opts,
  fallback_index,
  session_id,
  iteration_index,
) {
  let native_calls = llm_result?.native_tool_calls ?? []
  let parsed_calls = parsed?.calls ?? []
  let format = turn_opts?.tool_format ?? ""
  if format != "native" || len(native_calls) > 0 || len(parsed_calls) == 0 {
    return {triggered: false, accepted: false, fallback_index: fallback_index, calls: nil}
  }
  let new_index = fallback_index + 1
  let policy = turn_opts?.native_tool_fallback ?? "reject"
  let accepted = if policy == "allow" {
    true
  } else if policy == "allow_once" {
    new_index == 1
  } else {
    false
  }
  agent_record_native_tool_fallback(
    session_id,
    {
      iteration: iteration_index + 1,
      accepted: accepted,
      policy: policy,
      fallback_index: new_index,
      tool_call_count: len(parsed_calls),
    },
  )
  if !accepted {
    agent_session_inject_feedback(
      session_id,
      "native_tool_contract",
      __native_fallback_feedback(policy, new_index),
    )
  }
  let resolved_calls = if accepted {
    parsed_calls
  } else {
    []
  }
  return {triggered: true, accepted: accepted, fallback_index: new_index, calls: resolved_calls}
}

fn __resolve_max_concurrent_tools(turn_opts) {
  let raw = turn_opts?.max_concurrent_tools
  if type_of(raw) == "int" && raw > 1 {
    return raw
  }
  return 1
}

/**
 * Resolve the intra-turn failing-fan-out cap K (#A4). When a single model
 * response fans out a large batch of byte-identical FAILING tool calls, every
 * call is dispatched synchronously with no LLM call or progress check between
 * them, so the cross-turn loop-detector / no-progress terminator fire a whole
 * turn too late — after all N drain, having burned turn/wall budget and flooded
 * context with N identical errors (observed: 127 identical `edit` rejections in
 * ~2.7s on swift-feat). This cap is the intra-turn analog of the cross-turn
 * no-progress terminator: after the Kth consecutive byte-identical failing
 * result within ONE batch, the remaining identical calls are skipped and
 * collapsed into a single synthetic result.
 *
 * Default OFF: returns `0` (no cap) unless `intra_turn_failure_fanout_cap` is
 * set to a positive int. A value of e.g. `3` collapses after 3 identical
 * failures. Reachability is exercised by the `agent_loop_intra_turn_*`
 * conformance tests, which prove that flipping the flag changes dispatch
 * behavior end-to-end.
 */
fn __resolve_intra_turn_failure_fanout_cap(turn_opts) {
  let raw = turn_opts?.intra_turn_failure_fanout_cap
  if type_of(raw) == "int" && raw > 0 {
    return raw
  }
  return 0
}

fn __resolve_intra_turn_resource_fail_fast(turn_opts) -> bool {
  let raw = turn_opts?.intra_turn_resource_fail_fast
  if type_of(raw) == "bool" {
    return raw
  }
  return true
}

/**
 * Stable signature of a FAILING dispatch result, used to detect a fan-out of
 * byte-identical failing calls within one batch. Keyed on (tool_name,
 * args_hash, normalized failure text) so it is polyglot — it never inspects
 * language-specific content, only the tool identity, the exact arguments, and
 * the exact error/observation the tool returned. Returns `nil` for a successful
 * (or non-result) entry so successes never count toward the cap.
 */
fn __intra_turn_failure_signature(result) {
  if type_of(result) != "dict" {
    return nil
  }
  if __tool_result_ok(result) {
    return nil
  }
  let name = to_string(result?.tool_name ?? result?.name ?? "")
  let args = result?.arguments ?? {}
  let failure_text = to_string(result?.error ?? result?.observation ?? result?.rendered_result ?? result?.result ?? "")
  return sha256(json_stringify({name: name, args: args, failure: failure_text}))
}

/**
 * Stable signature of a tool CALL (before dispatch), keyed on (tool_name,
 * args). Used to skip the tail of a fan-out: once a streak of byte-identical
 * failing results trips the cap, every remaining call with this same call
 * signature is collapsed rather than executed. Polyglot — never inspects
 * language-specific content, only tool identity and exact arguments.
 */
fn __intra_turn_call_signature(call) {
  if type_of(call) != "dict" {
    return nil
  }
  let name = to_string(call?.name ?? call?.tool_name ?? "")
  let args = call?.arguments ?? {}
  return sha256(json_stringify({name: name, args: args}))
}

const __INTRA_TURN_RESOURCE_MUTATING_KINDS = ["edit", "write", "scaffold", "delete", "move", "mutation", "mutate"]

fn __intra_turn_call_args(call) {
  let raw = call?.arguments ?? call?.tool_args
  if type_of(raw) == "dict" {
    return raw
  }
  return {}
}

fn __intra_turn_annotation_text(value) -> string {
  return lowercase(trim(to_string(value ?? "")))
}

fn __intra_turn_annotations_mutate_resource(annotations) -> bool {
  if type_of(annotations) != "dict" {
    return false
  }
  let kind = __intra_turn_annotation_text(annotations?.kind ?? annotations?.tool_kind ?? annotations?.toolKind)
  if contains(__INTRA_TURN_RESOURCE_MUTATING_KINDS, kind) {
    return true
  }
  let side_effect = __intra_turn_annotation_text(
    annotations?.side_effect_level ?? annotations?.sideEffectLevel ?? annotations?.side_effect
      ?? annotations?.sideEffect,
  )
  if side_effect == "workspace_write" {
    return true
  }
  let mutation = __intra_turn_annotation_text(
    annotations?.mutation_classification ?? annotations?.mutationClassification
      ?? annotations?.mutation,
  )
  return mutation == "workspace_write"
}

fn __intra_turn_annotation_path_params(annotations) {
  if type_of(annotations) != "dict" {
    return []
  }
  let direct = annotations?.path_params ?? annotations?.pathParams ?? annotations?.resource_path_params
    ?? annotations?.resourcePathParams
  if type_of(direct) == "list" {
    return direct
  }
  if type_of(direct) == "string" {
    return [direct]
  }
  let schema = annotations?.arg_schema ?? annotations?.argSchema ?? {}
  let params = schema?.path_params ?? schema?.pathParams ?? []
  if type_of(params) == "list" {
    return params
  }
  if type_of(params) == "string" {
    return [params]
  }
  return []
}

fn __intra_turn_normalize_resource_path(raw) -> string {
  var path = trim(to_string(raw ?? ""))
  while starts_with(path, "./") {
    path = substring(path, 2)
  }
  return path
}

fn __intra_turn_push_resource_key(keys, value) {
  let path = __intra_turn_normalize_resource_path(value)
  if path == "" {
    return keys
  }
  let key = "workspace_path:" + path
  if contains(keys, key) {
    return keys
  }
  return keys.push(key)
}

fn __intra_turn_push_resource_value(keys, value) {
  if type_of(value) == "list" {
    var out = keys
    for item in value {
      out = __intra_turn_push_resource_key(out, item)
    }
    return out
  }
  return __intra_turn_push_resource_key(keys, value)
}

const __INTRA_TURN_FALLBACK_PATH_ARGS = [
  "path",
  "paths",
  "file",
  "files",
  "filepath",
  "file_path",
  "target_path",
  "source_path",
  "folder",
  "dir",
  "directory",
]

fn __intra_turn_fallback_resource_keys(args) {
  var keys = []
  for name in __INTRA_TURN_FALLBACK_PATH_ARGS {
    if args?[name] != nil {
      keys = __intra_turn_push_resource_value(keys, args[name])
    }
  }
  return keys
}

fn __intra_turn_resource_keys(call, tools, options = {}) {
  let tool_name = to_string(call?.name ?? call?.tool_name ?? "")
  let entry = __tool_registry_entry(tools, tool_name)
  let annotations = __tool_resource_annotations(entry, options?.policy, tool_name)
  if !__intra_turn_annotations_mutate_resource(annotations) {
    return []
  }
  let args = __intra_turn_call_args(call)
  var keys = []
  for param in __intra_turn_annotation_path_params(annotations) {
    let name = to_string(param)
    if name != "" && args?[name] != nil {
      keys = __intra_turn_push_resource_value(keys, args[name])
    }
  }
  if len(keys) == 0 {
    keys = __intra_turn_fallback_resource_keys(args)
  }
  return keys
}

fn __intra_turn_intersects(keys, failed_keys) -> bool {
  for key in keys {
    if contains(failed_keys, key) {
      return true
    }
  }
  return false
}

fn __intra_turn_add_failed_keys(failed_keys, keys) {
  var out = failed_keys
  for key in keys {
    if !contains(out, key) {
      out = out.push(key)
    }
  }
  return out
}

/**
 * Canonical NON-mutating write-failure vocabulary: failure shapes that a host
 * emits when the write was REJECTED BEFORE TOUCHING the file, so the on-disk
 * view is byte-identical to what the model already saw. Mirrors the same
 * canonical strings the product-error classifier (`__tool_result_write_product_error_text`)
 * already keys on, restricted to the pre-apply (no-mutation) subset. The
 * post-apply `## diagnostics (...)` shape is DELIBERATELY excluded — that edit
 * DID land, so a later same-resource sibling could be stale.
 */
fn __intra_turn_failure_text_is_nonmutating(text: string) -> bool {
  let lower = lowercase(trim(text))
  if lower == "" {
    return false
  }
  // Post-apply diagnostics mean the write LANDED — never treat as non-mutating.
  if lower.starts_with("## diagnostics (") || lower.contains("\n## diagnostics (") {
    return false
  }
  return lower.contains("[edit rejected]")
    || lower.contains("not applied")
    || lower.contains("old_string not found")
    || lower.starts_with("file already exists")
    || lower.starts_with("missing required parameter")
    || __tool_result_dispatch_rejection_text(lower)
}

/**
 * Tri-state mutation verdict for a FAILED tool result: did the failure leave the
 * workspace resource mutated?
 *
 *   `false` — the host reported the write was REJECTED before applying (a stale /
 *             missing anchor, "old_string not found", "[Edit rejected] ... NOT
 *             applied", an invalid-arguments/permission rejection). The view is
 *             byte-identical, so a later same-resource sibling is NOT stale.
 *   `nil`   — unknown. We cannot rule out a mutation (a post-apply diagnostic, an
 *             opaque error, a non-edit mutating tool), so the caller poisons the
 *             resource EXACTLY like today.
 *
 * Uses the SAME canonical write-failure vocabulary the dispatch classifier
 * (`__tool_result_write_product_error_text`) already relies on, so it does not
 * introduce a new, divergent text-matching surface. Conservative by
 * construction: anything not affirmatively a pre-apply rejection stays `nil`,
 * preserving today's safety property.
 */
fn __intra_turn_result_mutated(result) {
  if type_of(result) != "dict" {
    return nil
  }
  if !__tool_result_write_product_tool(result) {
    // Only reason about edit/scaffold/write/run_codemod rejection vocabulary.
    // Any other mutating tool stays unknown (poison), exactly as today.
    return nil
  }
  for key in ["error", "rendered_result", "observation", "result", "output"] {
    if result[key] != nil && __intra_turn_failure_text_is_nonmutating(to_string(result[key])) {
      return false
    }
  }
  return nil
}

/**
 * True when a failing same-resource call must POISON the resource for later
 * siblings — i.e. we cannot rule out that the failure left the resource mutated.
 * Preserves the safety property: poison unless the failure is an affirmatively
 * pre-apply (no-mutation) rejection.
 */
fn __intra_turn_failure_poisons_resource(result) -> bool {
  // Poison on `nil` (unknown — cannot rule out a mutation) and on any verdict
  // other than an affirmative `false`. Only a host-confirmed pre-apply rejection
  // (view unchanged) skips poisoning. Guards the safety property: unknown stays
  // conservative, exactly like today.
  let mutated = __intra_turn_result_mutated(result)
  if type_of(mutated) == "bool" && !mutated {
    return false
  }
  return true
}

fn __intra_turn_has_keyed_mutating_calls(tool_calls, tools, options = {}) -> bool {
  for call in tool_calls {
    if len(__intra_turn_resource_keys(call, tools, options)) > 0 {
      return true
    }
  }
  return false
}

/**
 * Synthetic collapsed result that stands in for the identical failing calls
 * skipped after the fan-out cap tripped. Mirrors the failing-result shape so
 * downstream rollups (rejected-tool tracking, history) treat it as one failed
 * call rather than executing N more.
 */
fn __intra_turn_collapsed_result(call, sample_result) {
  let name = to_string(call?.name ?? sample_result?.tool_name ?? "")
  let observation = "[collapsed remaining identical failing `" + name + "` call(s) "
    + "from this turn]\n"
    + "These calls had byte-identical arguments and produced the byte-identical "
    + "error already shown above, so they were NOT executed. Repeating the same "
    + "failing call cannot make progress — issue edits one at a time and inspect "
    + "each result, or change your approach (re-read the file / fix the argument "
    + "that was rejected) before retrying."
  return {
    ok: false,
    status: "collapsed_identical_failure",
    tool_name: name,
    tool_call_id: to_string(call?.id ?? call?.tool_call_id ?? ""),
    arguments: call?.arguments ?? {},
    result: nil,
    rendered_result: observation,
    observation: observation,
    error: observation,
    error_category: "intra_turn_failure_fanout_collapsed",
    executor: nil,
  }
}

fn __intra_turn_resource_blocked_result(call, keys) {
  let name = to_string(call?.name ?? call?.tool_name ?? "")
  let rendered_keys = join(keys, ", ")
  let observation = "[skipped dependent `" + name + "` call from this turn]\n"
    + "A previous mutating tool call in the same assistant response failed for "
    + "the same workspace resource ("
    + rendered_keys
    + "), so this call was NOT executed (its anchor may now be stale).\n"
    + "Recover in ONE next turn: (1) read the failure feedback from that earlier "
    + "call just above; (2) re-issue THIS `"
    + name
    + "` call now — re-anchor its "
    + "`old_string` against the file's CURRENT contents (look at the resource first "
    + "if the earlier call may have changed it). Do not wait extra turns and do not "
    + "re-send the call that already failed unchanged."
  return {
    ok: false,
    status: "blocked_by_prior_same_resource_failure",
    tool_name: name,
    tool_call_id: to_string(call?.id ?? call?.tool_call_id ?? ""),
    arguments: call?.arguments ?? {},
    result: nil,
    rendered_result: observation,
    observation: observation,
    error: observation,
    error_category: "intra_turn_resource_fail_fast",
    executor: nil,
  }
}

fn __callable(value) {
  let kind = type_of(value)
  return kind == "closure" || kind == "function" || kind == "fn"
}

fn __audit_flushes_from_result(result) {
  var flushes = []
  let single = result?._audit_flush
  if __callable(single) {
    flushes = flushes.push(single)
  }
  let many = result?._audit_flushes
  if type_of(many) == "list" {
    for flush in many {
      if __callable(flush) {
        flushes = flushes.push(flush)
      }
    }
  }
  return flushes
}

fn __collect_audit_flushes(dispatch) {
  var flushes = []
  for result in __dispatch_results_list(dispatch) {
    for flush in __audit_flushes_from_result(result) {
      flushes = flushes.push(flush)
    }
  }
  return flushes
}

fn __strip_internal_tool_result(result) {
  if type_of(result) != "dict" {
    return result
  }
  var clean = {}
  for key in result.keys() {
    if !starts_with(key, "_") {
      clean = clean + {[key]: result[key]}
    }
  }
  return clean
}

fn __strip_internal_dispatch(dispatch) {
  if type_of(dispatch) == "list" {
    var clean_results = []
    for result in dispatch {
      clean_results = clean_results.push(__strip_internal_tool_result(result))
    }
    return clean_results
  }
  if type_of(dispatch) != "dict" {
    return dispatch
  }
  var clean_dispatch = {}
  for key in dispatch.keys() {
    if !starts_with(key, "_") {
      clean_dispatch = clean_dispatch + {[key]: dispatch[key]}
    }
  }
  if type_of(dispatch?.results) == "list" {
    var clean_results = []
    for result in dispatch.results {
      clean_results = clean_results.push(__strip_internal_tool_result(result))
    }
    clean_dispatch = clean_dispatch + {results: clean_results}
  }
  return clean_dispatch
}

fn __spawn_audit_flushes(tasks, flushes) {
  var out = tasks
  for flush in flushes {
    let task = spawn {
      let _ = try {
        flush()
      }
    }
    out = out.push(task)
  }
  return out
}

fn __drain_audit_flushes(tasks) {
  for task in tasks {
    let _ = try {
      await(task)
    }
  }
}

fn __dispatch_tool_calls(session_id, tool_calls, turn_opts) {
  if len(tool_calls) == 0 {
    return {dispatch: nil, turn_opts: turn_opts, audit_flushes: []}
  }
  agent_tool_search_emit_queries(session_id, tool_calls, turn_opts)
  let tools = turn_opts?.tools
  let cap = __resolve_max_concurrent_tools(turn_opts)
  // `_stop_reason` is the turn's provider stop reason
  // (`stop`/`length`/`tool_calls`/...). The dispatch host primitive uses it
  // to cause-name empty-args tool calls: finish_reason=length means the
  // arguments were TRUNCATED by the output cap, anything else means the
  // provider dropped them (an IDE host bug report).
  let dispatch_options = {
    session_id: session_id,
    tool_format: turn_opts.tool_format,
    policy: turn_opts?.policy,
    approval_policy: turn_opts?.approval_policy,
    command_policy: turn_opts?.command_policy,
    permissions: turn_opts?.permissions,
    reminders: turn_opts?.reminders,
    _iteration: turn_opts?._iteration ?? 0,
    _tool_caller: turn_opts?._tool_caller,
    _max_concurrent_tools: cap,
    _prefetch_next_turn: turn_opts?.prefetch_next_turn ?? false,
    _stop_reason: turn_opts?._stop_reason ?? "",
    _intra_turn_failure_fanout_cap: __resolve_intra_turn_failure_fanout_cap(turn_opts),
    _intra_turn_resource_fail_fast: __resolve_intra_turn_resource_fail_fast(turn_opts),
  }
  let caller = turn_opts?._tool_caller
  let raw_dispatch = __dispatch_tool_calls_with_middleware(tool_calls, tools, dispatch_options, cap)
  let audit_flushes = __collect_audit_flushes(raw_dispatch)
  let dispatch = __strip_internal_dispatch(raw_dispatch)
  agent_session_record_tool_results(session_id, dispatch)
  return {
    dispatch: dispatch,
    turn_opts: agent_tool_search_record_results(session_id, tool_calls, dispatch, turn_opts),
    audit_flushes: audit_flushes,
  }
}

fn __invoke_tool_with_index(call, index, tools, options) {
  return __invoke_tool(call, tools, options + {_tool_call_index: index})
}

fn __dispatch_tool_calls_with_middleware(tool_calls, tools, options, cap) {
  // Middleware-enabled path. Each call invokes its own caller chain
  // inside a fresh closure scope, so `audit.layers` histories stay
  // independent across siblings. When `max_concurrent_tools > 1`,
  // dispatch siblings concurrently via `parallel settle` with the
  // requested cap; results come back in source order regardless of
  // completion order so text tool-call parsers that key on
  // declaration order still match.
  let resource_fail_fast = options?._intra_turn_resource_fail_fast ?? true
  let keyed_mutating_batch = resource_fail_fast
    && __intra_turn_has_keyed_mutating_calls(tool_calls, tools, options)
  if cap <= 1 || len(tool_calls) <= 1 || keyed_mutating_batch {
    // Intra-turn failing-fan-out cap (#A4). Dispatch serially; track the count
    // of consecutive byte-identical FAILING results. Once that streak reaches
    // the configured cap K, skip the remaining calls whose (tool_name, args)
    // signature matches the capped one — they would produce the same failure —
    // and substitute a single synthetic "collapsed" result. This is the
    // intra-turn analog of the cross-turn no-progress terminator. Default OFF
    // (`fanout_cap == 0`): the legacy "dispatch every call" behavior is
    // unchanged unless the flag is set. A success, or any failure/call with a
    // DIFFERENT signature, resets the streak, so a batch of distinct or
    // non-failing calls is never capped.
    let fanout_cap = options?._intra_turn_failure_fanout_cap ?? 0
    var results = []
    var streak_signature = nil
    var streak_count = 0
    var capped_call_signature = nil
    var capped_sample = nil
    var collapsed_emitted = false
    var failed_resource_keys = []
    for (index, call) in iter(tool_calls).enumerate() {
      let resource_keys = if resource_fail_fast {
        __intra_turn_resource_keys(call, tools, options)
      } else {
        []
      }
      if len(resource_keys) > 0 && __intra_turn_intersects(resource_keys, failed_resource_keys) {
        let blocked = __intra_turn_resource_blocked_result(call, resource_keys)
        __emit_synthetic_tool_lifecycle_finish(call, blocked, tools, options)
        results = results.push(blocked)
        continue
      }
      if fanout_cap > 0 && capped_call_signature != nil
        && __intra_turn_call_signature(call) == capped_call_signature {
        // Same failing call as the one that tripped the cap — skip dispatch.
        if !collapsed_emitted {
          let collapsed = __intra_turn_collapsed_result(call, capped_sample)
          __emit_synthetic_tool_lifecycle_finish(call, collapsed, tools, options)
          results = results.push(collapsed)
          collapsed_emitted = true
        }
        continue
      }
      let result = __invoke_tool_with_index(call, index, tools, options)
      results = results.push(result)
      // Poison this resource for later same-resource siblings only when the
      // failure could have left it mutated. A host-confirmed no-op failure
      // (rejected anchor / "old_string not found") leaves the view byte-identical,
      // so an independent later edit to the same file is NOT stale and may run.
      // Unknown (no structured signal) still poisons, preserving today's behavior.
      if len(resource_keys) > 0 && !__tool_result_ok(result)
        && __intra_turn_failure_poisons_resource(result) {
        failed_resource_keys = __intra_turn_add_failed_keys(failed_resource_keys, resource_keys)
      }
      if fanout_cap > 0 {
        let signature = __intra_turn_failure_signature(result)
        if signature == nil {
          streak_signature = nil
          streak_count = 0
        } else if signature == streak_signature {
          streak_count = streak_count + 1
          if streak_count >= fanout_cap {
            let new_capped = __intra_turn_call_signature(call)
            // A DISTINCT fan-out group tripped the cap: reset the
            // collapse-emitted latch so this group gets its OWN single
            // synthetic "collapsed" result. Without this reset the latch
            // (set by an earlier group) suppresses every collapse marker
            // after the first, silently dropping the tail of later groups
            // from `results` with no entry at all.
            if new_capped != capped_call_signature {
              collapsed_emitted = false
            }
            capped_call_signature = new_capped
            capped_sample = result
          }
        } else {
          streak_signature = signature
          streak_count = 1
        }
      }
    }
    return results
  }
  var indexed = []
  for (index, call) in iter(tool_calls).enumerate() {
    indexed = indexed.push({index: index, call: call})
  }
  let settled = parallel settle indexed with { max_concurrent: cap } { entry ->
    __invoke_tool_with_index(entry.call, entry.index, tools, options)
  }
  var results = []
  for r in settled.results {
    if is_ok(r) {
      results = results.push(unwrap(r))
    } else {
      // `__invoke_tool` traps its own middleware exceptions, so a thrown
      // value here is a VM-level bug (e.g. parallel-task plumbing). Surface
      // it as a synthetic error result rather than tear down the loop.
      let err = unwrap_err(r)
      if error_category(err) == "cancelled" {
        throw err
      }
      results = results
        .push(
        {
          ok: false,
          status: "error",
          tool_name: "",
          tool_call_id: "",
          arguments: {},
          result: nil,
          rendered_result: to_string(err),
          observation: to_string(err),
          error: to_string(err),
          error_category: "tool_parallel_dispatch_exception",
          executor: nil,
        },
      )
    }
  }
  return results
}

fn __sync_tool_search_state(opts, turn_opts) {
  if turn_opts?._tool_search_client == nil {
    return opts
  }
  return opts + {_tool_search_client: turn_opts._tool_search_client}
}

fn __dispatch_results_list(dispatch) {
  if dispatch == nil {
    return []
  }
  if type_of(dispatch) == "list" {
    return dispatch
  }
  return dispatch?.results ?? []
}

fn __tool_result_ok(result) {
  if __tool_result_product_error(result) {
    return false
  }
  if result?.ok != nil {
    return result.ok ? true : false
  }
  if result?.success != nil {
    return result.success ? true : false
  }
  let status = result?.status ?? ""
  return status == "ok" || status == "success"
}

/**
 * Some host tools report product-level rejections as a successful transport
 * result whose rendered/observation text begins with `Error:`. Treating those
 * as successful makes same-turn resource fail-fast and rejected-tool accounting
 * blind to edit rejections. Write tools can also report in-band failures with
 * Burin's canonical edit/scaffold vocabulary (`## Diagnostics (...)`,
 * `[Edit rejected]`, missing path, etc.). This predicate intentionally does not
 * rewrite the result payload; it only classifies the result for
 * scheduling/lifecycle.
 */
const __PRODUCT_ERROR_WRITE_TOOL_NAMES = ["edit", "scaffold", "run_codemod", "write"]

fn __tool_result_write_product_tool(result) -> bool {
  let name = lowercase(trim(to_string(result?.tool_name ?? result?.name ?? result?.tool_name_raw ?? "")))
  return contains(__PRODUCT_ERROR_WRITE_TOOL_NAMES, name)
}

fn __tool_result_dispatch_rejection_text(lower: string) -> bool {
  return lower.contains("\"error\":\"invalid_arguments\"")
    || lower.contains("\"error\": \"invalid_arguments\"")
    || lower.contains("error: invalid_arguments")
    || lower.contains("\"error\":\"permission_denied\"")
    || lower.contains("\"error\": \"permission_denied\"")
    || lower.contains("error: permission_denied")
}

fn __tool_result_write_product_error_text(text: string) -> bool {
  let lower = lowercase(trim(text))
  if lower == "" {
    return false
  }
  if lower.starts_with("## diagnostics (") {
    return !lower.starts_with("## diagnostics (0 errors")
  }
  if lower.starts_with("[result of ") && lower.contains("\n## diagnostics (") {
    return !lower.contains("\n## diagnostics (0 errors")
  }
  return lower.starts_with("no ")
    || lower.starts_with("file already exists")
    || lower.starts_with("missing required parameter")
    || lower.contains("[edit rejected]")
    || __tool_result_dispatch_rejection_text(lower)
}

fn __tool_result_product_error(result) -> bool {
  if type_of(result) != "dict" {
    return false
  }
  let write_tool = __tool_result_write_product_tool(result)
  for key in ["error", "observation", "rendered_result", "result", "output"] {
    if result[key] != nil {
      let text = trim(to_string(result[key]))
      if starts_with(text, "Error:") {
        return true
      }
      if starts_with(text, "[result of ") && contains(text, "\nError:") {
        return true
      }
      if write_tool && __tool_result_write_product_error_text(text) {
        return true
      }
    }
  }
  return false
}

fn __tool_result_name(result) {
  return result?.tool_name ?? result?.name ?? ""
}

fn __tool_names_by_status(dispatch, want_ok) {
  let results = __dispatch_results_list(dispatch)
  var names = []
  for result in results {
    let name = __tool_result_name(result)
    if name != "" && __tool_result_ok(result) == want_ok {
      names = names.push(name)
    }
  }
  return names
}

// Workspace-mutating annotation kinds (mirrors postturn's tool-surface
// classifier vocabulary). A successful call to a tool of one of these kinds is
// a corrective edit for the current-failure model's `write_epoch`.
const __REPAIR_EDIT_KINDS = ["edit", "write", "scaffold", "delete", "mutation", "mutate"]

/**
 * __turn_made_edit reports whether `dispatch` (a single turn's dispatch
 * results) contains a SUCCESSFUL workspace-mutating tool call, by looking up
 * each successful tool's annotation kind in the registry. Reuses
 * `__tool_names_by_status` + `__tool_registry_entry`; only consulted when
 * post-edit verification or repair-aware diagnostics are enabled.
 */
fn __turn_made_edit(dispatch, tools) -> bool {
  if dispatch == nil {
    return false
  }
  for name in __tool_names_by_status(dispatch, true) {
    let entry = __tool_registry_entry(tools, name)
    let annotations = __tool_entry_annotations(entry)
    let kind = lowercase(to_string(annotations?.kind ?? annotations?.tool_kind ?? annotations?.toolKind ?? ""))
    if contains(__REPAIR_EDIT_KINDS, kind) {
      return true
    }
  }
  return false
}

fn __agent_loop_verify_closure_exists(turn_opts) -> bool {
  return turn_opts?.verify_completion != nil || turn_opts?.verify_completion_judge != nil
}

fn __agent_loop_post_edit_reverify_mandated(repair_cfg, turn_opts, stall_state, dispatch, tools) -> bool {
  let has_live_failure = stall_state.last_diagnostic_class == "fail" || stall_state.reverify_owed
  return repair_cfg.post_edit_reverify
    && __agent_loop_verify_closure_exists(turn_opts)
    && has_live_failure
    && __turn_made_edit(dispatch, tools)
}

// Terminal statuses the reserved-verify guard may intercept: a run that ran out
// of iteration/wall-clock/cost runway (`budget_exhausted`) or was stopped as a
// thrash (`stuck`). Other terminals (suspended / verify_capped / scope_alert /
// done) are deliberate and must not be reinterpreted.
const __RESERVED_VERIFY_TERMINAL_STATUSES = ["budget_exhausted", "stuck"]

/**
 * __agent_loop_should_spend_reserve decides whether the reserved terminal-verify
 * guard should run a final verify(+repair) before the run terminates. It fires
 * ONLY when the guard is enabled, the run is terminating on a budget/stuck
 * boundary, the transcript still has an UNVERIFIED source write (the model
 * edited then ran out of runway without a passing verification), a
 * `verify_completion`/`verify_completion_judge` closure exists to run, and the
 * shared verify-attempt cap has not been hit. Default OFF: when
 * `reserved_terminal_verify` is false this always returns false, so behavior is
 * byte-identical to today. The `write_unverified` signal REUSES the existing
 * post-edit re-verify bookkeeping (it is true whenever a turn made a workspace
 * edit that no passing verification has cleared), the same notion the in-loop
 * `reverify_owed` mandate keys on. This is the budget-exit counterpart to the
 * loop-cap `iteration >= current_max` done-conversion (#3629), which only fires
 * when an in-loop reverify ALREADY confirmed green; this guard runs the verifier
 * on the terminal budget/stuck break where NO reverify fired. The guard verifies
 * first and only spends a reserve iteration on a repair turn when the verify
 * comes back red and reserve remains, so this predicate gates the verify.
 */
fn __agent_loop_should_spend_reserve(
  repair_cfg,
  turn_opts,
  final_status,
  write_unverified: bool,
  verify_allowed: bool,
) -> bool {
  return repair_cfg.reserved_terminal_verify
    && contains(__RESERVED_VERIFY_TERMINAL_STATUSES, final_status)
    && write_unverified
    && verify_allowed
    && __agent_loop_verify_closure_exists(turn_opts)
}

fn __merge_tool_names(existing, additions) {
  var merged = existing ?? []
  let values = additions ?? []
  for name in values {
    if name != "" && !contains(merged, name) {
      merged = merged.push(name)
    }
  }
  return merged
}

fn __merge_hook_dict(base, patch, label) {
  if patch == nil {
    return base
  }
  if type_of(patch) != "dict" {
    throw "agent_loop: post_turn_callback `" + label + "` must be a dict"
  }
  return base + patch
}

fn __strip_internal_keys(patch) {
  if patch == nil {
    return patch
  }
  if type_of(patch) != "dict" {
    return patch
  }
  var clean = {}
  for key in patch.keys() {
    if !starts_with(key, "_") {
      clean = clean + {[key]: patch[key]}
    }
  }
  return clean
}

fn __apply_post_turn_options(opts, outcome) {
  var updated = opts
  let next_patch = __strip_internal_keys(outcome?.next_options)
  updated = __merge_hook_dict(updated, next_patch, "next_options")
  let narrowing = outcome?.narrowing
  if narrowing != nil {
    updated = updated
      + {
      _tool_surface_narrowing_history: narrowing?.history ?? [],
      _tool_surface_narrowed_tools: narrowing?.narrowed_tools,
    }
  }
  let llm_patch = outcome?.llm_options
  if llm_patch != nil {
    if type_of(llm_patch) != "dict" {
      throw "agent_loop: post_turn_callback `llm_options` must be a dict"
    }
    let base_llm_options = updated?.llm_options ?? {}
    updated = updated + {llm_options: base_llm_options + llm_patch}
  }
  return updated
}

fn __agent_loop_llm_overrides(opts) {
  let overrides = opts?.llm_options
  if type_of(overrides) == "dict" {
    return overrides
  }
  return {}
}

fn __agent_loop_effective_llm_options(opts) {
  let base = opts ?? {}
  return agent_sanitize_model_options(base + __agent_loop_llm_overrides(base))
}

fn __agent_loop_with_llm_render_context(opts, render) {
  let effective = __agent_loop_effective_llm_options(opts)
  let provider = to_string(effective?.provider ?? "")
  let model = to_string(effective?.model ?? "")
  let pushed = __push_llm_render_context(provider, model)
  defer {
    if pushed {
      __pop_llm_render_context()
    }
  }
  return render(effective)
}

fn __agent_loop_build_turn_prompt(session, opts, iteration) {
  return __agent_loop_with_llm_render_context(
    opts,
    { effective ->
      let fragments = agent_build_turn_system_fragments(session, effective, iteration)
      var parts = []
      for fragment in fragments {
        parts = parts.push(fragment.body)
      }
      return {
        fragments: fragments,
        system: join(parts, "\n\n"),
        messages: agent_build_turn_messages(session, effective, iteration),
      }
    },
  )
}

fn __next_text_only_count(tool_count, consecutive_text_only) {
  if tool_count == 0 {
    return consecutive_text_only + 1
  }
  return 0
}

/**
 * "Turns since meaningful progress" — a decaying churn signal that, unlike
 * `__next_text_only_count`, does NOT fully reset on a single tool dispatch.
 * Meaningful progress = at least one SUCCESSFUL tool this turn (a mutating
 * or verify call that actually applied), not a rejected/no-op one. A turn
 * with no successful tool increments; a turn that made progress decays the
 * streak by one. This keeps mixed prose+tool churn (a successful call every
 * few turns interleaved with toolless narration) from evading the
 * escalating progress nudge the way the zeroing counter does.
 */
fn __next_progress_count(made_progress, turns_since_progress) {
  if made_progress {
    return if turns_since_progress > 0 {
      turns_since_progress - 1
    } else {
      0
    }
  }
  return turns_since_progress + 1
}

/**
 * Escalating, depth-keyed copy for the toolless/no-progress churn nudge
 * (M1-1). Bounded: `__agent_loop_run` only invokes this below the hard
 * `max_nudges` stop, so it never nudges forever, and only when the
 * content-specific text-mode nudge (fence / missing-call recovery) did NOT fire
 * this turn — the content nudge takes precedence so a turn is never
 * double-injected; this streak nudge is the fallback for toolless churn.
 * Depth is `turns_since_progress`.
 *
 * @effects: []
 * @errors: []
 * @api_stability: experimental
 */
pub fn __progress_nudge_text(depth, has_tools, tool_mode = "text", made_tool_calls = false) {
  let action = if has_tools {
    if tool_mode == "native" {
      "Call exactly one native tool through the provider tool channel now, or give your final answer and stop."
    } else {
      "Emit exactly one well-formed <tool_call> now, or state your final answer in <user_response> and stop."
    }
  } else {
    "Make concrete progress in your reply now, or state your final answer and stop."
  }
  if depth >= 4 {
    // The "stuck in a narration loop / Do not restate the plan" framing is only
    // truthful when the recent turns were PURE PROSE. When the model IS issuing
    // real tool calls (edits/looks/searches/runs) but the loop's progress signal
    // still has not advanced, accusing it of narrating is wrong and unhelpful —
    // it is making changes that are not landing. Steer it to a DIFFERENT change
    // instead of telling it to stop narrating.
    if made_tool_calls {
      return "You have made no progress for "
        + to_string(depth)
        + " turns — your tool calls are not advancing the task. Stop repeating the same kind of change: re-read the failing output, target a DIFFERENT location, and make one concrete change. "
        + action
    }
    return "You have made no progress for "
      + to_string(depth)
      + " turns — you are stuck in a narration loop. "
      + action
      + " Do not restate the plan."
  }
  if depth >= 3 {
    return "Still no progress after "
      + to_string(depth)
      + " turns. "
      + action
  }
  return "No progress last turn. " + action
}

/**
 * Build the `iteration_info` payload for the `iteration_end` event from
 * the LLM result plus the loop's per-iteration aggregates. Carries `provider`,
 * `model`, `response_ms`, token counts, and `thinking_chars` so live
 * pulse-check consumers (fleet hooks, ACP clients) can attribute
 * latency and surface "still working" indicators without re-parsing
 * the transcript JSONL. Empty/missing fields are dropped so the event
 * stays small for providers that don't report telemetry.
 */
fn __iteration_info_payload(llm_result, tool_count, visible_text, aggregates = nil) {
  let telemetry = llm_result?.provider_telemetry ?? {}
  let thinking = llm_result?.thinking ?? ""
  let thinking_chars = if type_of(thinking) == "string" {
    len(thinking)
  } else {
    0
  }
  let extra = aggregates ?? {}
  return {
    tool_count: tool_count,
    text: visible_text,
    provider: llm_result?.provider ?? "",
    model: llm_result?.model ?? "",
    response_ms: telemetry?.client_wall_ms ?? 0,
    input_tokens: llm_result?.input_tokens ?? 0,
    output_tokens: llm_result?.output_tokens ?? 0,
    thinking_chars: thinking_chars,
  }
    + extra
}

fn __agent_loop_elapsed_ms(start_ms) {
  let elapsed = __agent_loop_clock_now_ms() - start_ms
  if elapsed < 0 {
    return 0
  }
  return elapsed
}

fn __agent_loop_cost_usd(totals) {
  let value = to_float(totals?.cost_usd ?? 0.0)
  if value == nil {
    return 0.0
  }
  return value
}

fn __agent_loop_budget_aggregates(session_id, totals, loop_start_ms) {
  let resolved_totals = totals ?? agent_session_totals(session_id)
  return {
    cost_usd: __agent_loop_cost_usd(resolved_totals),
    wall_clock_ms: __agent_loop_elapsed_ms(loop_start_ms),
  }
}

fn __agent_loop_iteration_info(
  session_id,
  llm_result,
  tool_count,
  visible_text,
  totals,
  loop_start_ms,
) {
  return __iteration_info_payload(
    llm_result,
    tool_count,
    visible_text,
    __agent_loop_budget_aggregates(session_id, totals, loop_start_ms),
  )
}

fn __agent_loop_budget_exhaustion(
  session_id,
  budget,
  iteration,
  totals,
  loop_start_ms,
  max_iterations,
) {
  let aggregates = __agent_loop_budget_aggregates(session_id, totals, loop_start_ms)
  if budget?.wall_clock_ms != nil && aggregates.wall_clock_ms >= budget.wall_clock_ms {
    return aggregates
      + {exhausted: true, kind: "wall_clock", iteration: iteration, max_iterations: max_iterations}
  }
  if budget?.total_cost_usd != nil && aggregates.cost_usd >= budget.total_cost_usd {
    return aggregates
      + {exhausted: true, kind: "total_cost", iteration: iteration, max_iterations: max_iterations}
  }
  return aggregates + {exhausted: false, kind: "", iteration: iteration, max_iterations: max_iterations}
}

fn __agent_loop_emit_budget_exhausted(session_id, exhaustion) {
  agent_emit_event(
    session_id,
    "budget_exhausted",
    {
      kind: exhaustion?.kind ?? "budget_exhausted",
      max_iterations: exhaustion?.max_iterations ?? 0,
      iteration: exhaustion?.iteration ?? 0,
      cost_usd: exhaustion?.cost_usd ?? 0.0,
      wall_clock_ms: exhaustion?.wall_clock_ms ?? 0,
    },
  )
}

fn __agent_loop_record_budget_stop(decisions, iteration, current_max, reason) {
  return decisions
    .push(
    {
      iteration: iteration,
      action: "stop",
      old_limit: current_max,
      new_limit: current_max,
      reason: reason,
      status: "budget_exhausted",
    },
  )
}

fn __agent_loop_record_terminal_callback_continue(decisions, iteration, old_limit, new_limit) {
  return decisions
    .push(
    {
      iteration: iteration,
      action: "extend",
      old_limit: old_limit,
      new_limit: new_limit,
      reason: "terminal_callback_continue",
      status: "",
    },
  )
}

fn __agent_loop_state_budget_fields(
  session_id,
  totals,
  loop_start_ms,
  budget,
  consecutive_failure_count,
) {
  let aggregates = __agent_loop_budget_aggregates(session_id, totals, loop_start_ms)
  let failure_config = __agent_loop_consecutive_failure_config(budget)
  return {
    wall_clock_ms: aggregates.wall_clock_ms,
    wall_clock_limit_ms: budget?.wall_clock_ms,
    cost_usd: aggregates.cost_usd,
    total_cost_limit_usd: budget?.total_cost_usd,
    consecutive_failures: consecutive_failure_count,
    consecutive_failure_limit: failure_config?.max,
  }
}

fn __agent_loop_failure_matches_kind(error, wanted) {
  let category = to_string(error?.category ?? "")
  let reason = to_string(error?.reason ?? "")
  let kind = to_string(error?.kind ?? "")
  let status = to_int(error?.status ?? error?.status_code ?? 0) ?? 0
  if wanted == "transient" {
    return kind == "transient" || category == "transient_network" || category == "timeout"
      || reason == "timeout"
  }
  if wanted == "rate_limit" {
    return kind == "rate_limit" || category == "rate_limit" || category == "rate_limited"
      || reason == "rate_limit"
      || status == 429
  }
  if wanted == "provider_5xx" {
    return kind == "provider_5xx" || category == "server_error" || category == "overloaded"
      || reason == "server_error"
      || (status >= 500 && status < 600)
  }
  return category == wanted || reason == wanted || kind == wanted
}

fn __agent_loop_consecutive_failure_config(budget) {
  let config = budget?.consecutive_failures
  if type_of(config) != "dict" {
    return nil
  }
  return config
}

fn __agent_loop_tracks_failure(error, config) {
  if config == nil {
    return false
  }
  for kind in config?.kinds ?? ["transient", "rate_limit", "provider_5xx"] {
    if __agent_loop_failure_matches_kind(error, kind) {
      return true
    }
  }
  return false
}

fn __agent_loop_is_escalation_transport_failure(error) {
  let status = to_int(error?.status ?? error?.status_code ?? 0) ?? 0
  if status >= 400 && status < 500 && status != 429 {
    return false
  }
  let category = to_string(error?.category ?? "")
  let reason = to_string(error?.reason ?? "")
  let kind = to_string(error?.kind ?? "")
  if category == "circuit_open" || reason == "circuit_open" || kind == "circuit_open" {
    return true
  }
  return __agent_loop_failure_matches_kind(error, "transient")
    || __agent_loop_failure_matches_kind(error, "rate_limit")
    || __agent_loop_failure_matches_kind(error, "provider_5xx")
}

fn __agent_loop_transport_abort_error(error) {
  if type_of(error) == "dict" {
    return error + {escalation_aborted_provider_transport: true}
  }
  return {message: to_string(error), escalation_aborted_provider_transport: true}
}

/**
 * Surface a NON-tracked provider failure as an observable event instead of a
 * silent `break`. Mirrors the tracked-failure branch's `iteration_end` event
 * so any provider_error (including a fast pre-dispatch escalation failure) is
 * always visible in the transcript, carrying the provider/model/status/error
 * that caused it. The dispatch_skipped/skip_reason shape matches the tracked
 * branch so existing consumers parse it uniformly.
 */
fn __agent_loop_emit_provider_error(
  session_id,
  iteration_index,
  call,
  llm_opts,
  loop_start_ms,
  escalation_fallback,
  skip_reason_override,
) {
  let aggregates = __agent_loop_budget_aggregates(session_id, nil, loop_start_ms)
  let skip_reason = if skip_reason_override != "" {
    skip_reason_override
  } else if escalation_fallback {
    "escalation_provider_failure"
  } else {
    "provider_failure"
  }
  agent_emit_event(
    session_id,
    "iteration_end",
    {
      iteration: iteration_index + 1,
      iteration_info: aggregates
        + {
        dispatch_skipped: true,
        skip_reason: skip_reason,
        provider_error: call?.error ?? {},
        provider: to_string(llm_opts?.provider ?? ""),
        model: to_string(llm_opts?.model ?? ""),
        provider_status: to_string(call?.status ?? ""),
        provider_error_message: to_string(call?.error?.message ?? call?.error ?? ""),
        consecutive_failures: 0,
        escalation_fallback: escalation_fallback,
        escalation_transport_abort: skip_reason == "escalation_aborted_provider_transport",
      },
    },
  )
}

fn __agent_loop_finalize_failed(session, iteration) {
  try {
    agent_session_finalize(
      session.session_id,
      {final_status: "failed", stop_reason: "error", iterations: iteration},
    )
  } catch (e) {
  }
}

fn __agent_loop_current_worker() {
  let ctx = runtime_context()
  let worker_id = ctx?.worker_id
  if type_of(worker_id) != "string" || worker_id == "" {
    return nil
  }
  for worker in list_agents() {
    if worker?.id == worker_id {
      return worker
    }
  }
  return nil
}

fn __agent_loop_suspend_reminder_body(reason) {
  if type_of(reason) == "string" && reason != "" {
    return "Worker suspended before the next turn: " + reason
  }
  return "Worker suspended before the next turn."
}

fn __agent_loop_suspend_initiator(value) {
  let text = to_string(value ?? "operator")
  if text == "self_initiated" {
    return "self"
  }
  return text
}

fn __agent_loop_suspend_checkpoint(session, iteration) {
  let worker = __agent_loop_current_worker()
  if worker == nil || worker?.status != "suspended" {
    return nil
  }
  let suspension = worker?.suspension ?? {}
  let reason = to_string(suspension?.reason ?? "")
  let initiator = __agent_loop_suspend_initiator(suspension?.initiator)
  let conditions = suspension?.conditions
  let payload = {
    status: "suspended",
    handle: worker,
    worker: worker,
    reason: reason,
    initiator: initiator,
    conditions: conditions,
    iterations_completed: iteration,
    session_id: session.session_id,
  }
  agent_session_inject(
    session.session_id,
    transcript_reminder_event(
      {
        body: __agent_loop_suspend_reminder_body(reason),
        source: "in_pipeline",
        tags: ["agent_loop", "worker_suspend"],
        dedupe_key: "worker_suspend:" + worker.id,
        ttl_turns: 1,
        fired_at_turn: iteration + 1,
      },
    ),
  )
  return payload
}

/**
 * Mode filter table for `__agent_loop_checkpoint`. Each row encodes the
 * invariant for one seam: which bridge modes are eligible to drain, and
 * whether the host should pull the agent_inbox feedback queue.
 *
 *   iteration_start, post_tool_dispatch, iteration_end → drain `interrupt_immediate`
 *     and `finish_step`; the model will see whatever lands on its next
 *     prompt build, so both modes get the same opportunity here.
 *   pre_tool_dispatch → drain `interrupt_immediate` only. This is the
 *     "stop before the tool fires" seam — `finish_step` semantics
 *     would defeat the point (it means "after the current tool batch").
 *     If anything arrives, the caller skips the pending tool batch.
 *   pre_compact, post_compact → bracket the compactor with an
 *     agent_inbox drain so async producers (tool completions, MCP
 *     notifications, command-policy feedback) land in the transcript
 *     before the summarizer sees it and the next prompt is built.
 *   daemon_idle_pre, daemon_idle_post → drain `interrupt_immediate`
 *     only; the daemon path doesn't queue `finish_step`-mode injections.
 *   loop_exit → drain `audit_only`. The other two modes were already
 *     drained earlier in the loop body. `audit_only` reminders land in
 *     the transcript at this seam but are NEVER rendered into a model
 *     prompt — no further LLM call runs after `loop_exit` (harn#2212).
 *     Hosts that need the model to see a reminder before the agent
 *     terminates must use `finish_step` (drained at every iteration
 *     boundary, including the last `iteration_end` before the loop breaks).
 */
fn __agent_loop_checkpoint_modes(kind) {
  if kind == "pre_tool_dispatch" || kind == "daemon_idle_pre" || kind == "daemon_idle_post" {
    return {immediate: true, finish_step: false, audit_only: false, inbox: false}
  }
  if kind == "iteration_start" || kind == "post_tool_dispatch" || kind == "iteration_end" {
    return {immediate: true, finish_step: true, audit_only: false, inbox: false}
  }
  if kind == "pre_compact" || kind == "post_compact" {
    return {immediate: false, finish_step: false, audit_only: false, inbox: true}
  }
  if kind == "loop_exit" {
    return {immediate: false, finish_step: false, audit_only: true, inbox: false}
  }
  return {immediate: false, finish_step: false, audit_only: false, inbox: false}
}

/**
 * Single source of truth for "the loop is at a safe injection seam."
 * Every drain site in the agent loop body and the daemon idle path
 * routes through here so plugin authors and replayers see one canonical
 * event (`LoopCheckpoint`) instead of having to enumerate inline calls.
 *
 * Returns a result dict carrying `delivered` (bridge injections drained
 * at this seam), `inbox_delivered` (inbox feedback notes drained), and
 * `dispatch_skipped` (`pre_tool_dispatch` short-circuit: an
 * `interrupt_immediate` arrival here means the pending tool batch is
 * cancelled and the loop iterates once more so the LLM sees the
 * injection before the tool would have fired). Callers branch on
 * `delivered` for the "continue if a steer arrived" behavior the
 * stalled-done-judge and post-turn paths rely on.
 */
fn __agent_loop_checkpoint(session_id, kind, opts = nil) {
  let modes = __agent_loop_checkpoint_modes(kind)
  var delivered = 0
  if modes.immediate {
    let result = agent_session_drain_bridge_injections(session_id, "interrupt_immediate")
    delivered = delivered + (result?.delivered ?? 0)
  }
  let immediate_count = delivered
  if modes.finish_step {
    let result = agent_session_drain_bridge_injections(session_id, "finish_step")
    delivered = delivered + (result?.delivered ?? 0)
  }
  if modes.audit_only {
    let result = agent_session_drain_bridge_injections(session_id, "audit_only")
    delivered = delivered + (result?.delivered ?? 0)
  }
  var inbox_delivered = 0
  if modes.inbox {
    let pending = agent_session_drain_feedback(session_id)
    for note in pending {
      agent_session_inject_feedback(session_id, note.kind, note.content)
      inbox_delivered = inbox_delivered + 1
    }
  }
  let dispatch_skipped = kind == "pre_tool_dispatch" && immediate_count > 0
  let iteration = to_int(opts?.iteration ?? 0)
  agent_emit_event(
    session_id,
    "loop_checkpoint",
    {
      iteration: iteration,
      kind: kind,
      delivered: delivered,
      inbox_delivered: inbox_delivered,
      dispatch_skipped: dispatch_skipped,
    },
  )
  __host_fire_session_hook(
    "loop_checkpoint",
    {
      session_id: session_id,
      iteration: iteration,
      kind: kind,
      delivered: delivered,
      inbox_delivered: inbox_delivered,
      dispatch_skipped: dispatch_skipped,
    },
  )
  return {
    delivered: delivered,
    inbox_delivered: inbox_delivered,
    dispatch_skipped: dispatch_skipped,
    kind: kind,
  }
}

fn __agent_loop_fire_resume_continuity(session, opts) {
  let payload = opts?._resume_continuity
  if type_of(payload) != "dict" {
    return opts
  }
  let _ = agent_reminder_providers_fire(
    session.session_id,
    "worker_resumed",
    payload
      + {
      session_id: session.session_id,
      session: {id: session.session_id},
      turn: payload?.turn ?? 0,
      iteration: payload?.iteration ?? 0,
    },
    opts,
  )
  return omit(opts, ["_resume_continuity"])
}

fn __scope_classifier_recent_context(messages, limit) {
  if type_of(messages) != "list" {
    return []
  }
  let cap = if type_of(limit) == "int" && limit > 0 {
    limit
  } else {
    3
  }
  let total = len(messages)
  var start = total - cap
  if start < 0 {
    start = 0
  }
  var out = []
  var i = start
  while i < total {
    let msg = messages[i]
    out = out
      .push({role: to_string(msg?.role ?? ""), content: msg?.content ?? msg?.text ?? ""})
    i = i + 1
  }
  return out
}

fn __scope_classifier_confidence(value, fallback) {
  let parsed = to_float(value ?? fallback)
  if parsed == nil {
    return fallback
  }
  if parsed < 0.0 {
    return 0.0
  }
  if parsed > 1.0 {
    return 1.0
  }
  return parsed
}

fn __scope_classifier_label(value) {
  let label = lowercase(trim(to_string(value ?? "")))
  if label == "in_scope" || label == "inscope" || label == "in-scope" {
    return "in_scope"
  }
  if label == "out_of_scope" || label == "outscope" || label == "out-of-scope" {
    return "out_of_scope"
  }
  if label == "escalate" || label == "ambiguous" || label == "uncertain" {
    return "escalate"
  }
  return "escalate"
}

fn __scope_classifier_normalize(raw, session_id, iteration) {
  if raw == nil {
    return nil
  }
  if type_of(raw) != "dict" {
    return {
      label: "escalate",
      original_label: "invalid",
      confidence: 0.0,
      confidence_threshold: 0.65,
      evidence: "scope classifier returned " + type_of(raw) + ", not a dict",
      session_id: session_id,
      iteration: iteration,
      skip_main_turn: false,
    }
  }
  let threshold = __scope_classifier_confidence(raw?.confidence_threshold ?? raw?.threshold, 0.65)
  let original_label = __scope_classifier_label(raw?.original_label ?? raw?.label)
  let confidence = __scope_classifier_confidence(raw?.confidence, 0.0)
  let label = if original_label != "escalate" && confidence < threshold {
    "escalate"
  } else {
    original_label
  }
  let evidence = trim(to_string(raw?.evidence ?? raw?.reason ?? raw?.reasoning ?? ""))
  return raw
    + {
    label: label,
    original_label: original_label,
    confidence: confidence,
    confidence_threshold: threshold,
    evidence: if evidence == "" {
      "no evidence provided"
    } else {
      evidence
    },
    session_id: session_id,
    iteration: iteration,
    skip_main_turn: raw?.skip_main_turn ?? true,
  }
}

fn __scope_classifier_fail_open(session_id, iteration, err) {
  return {
    label: "escalate",
    original_label: "error",
    confidence: 0.0,
    confidence_threshold: 0.65,
    evidence: "scope classifier failed: " + to_string(err),
    error: to_string(err),
    session_id: session_id,
    iteration: iteration,
    skip_main_turn: false,
  }
}

fn __run_pre_turn_scope_classifier(classifier, session, message, turn_opts, iteration_index) {
  if classifier == nil {
    return nil
  }
  let iteration = iteration_index + 1
  let messages = agent_session_messages(session.session_id)
  let anchor = agent_session_workspace_anchor(session.session_id)
  let payload = {
    session_id: session.session_id,
    session: {id: session.session_id},
    iteration: iteration,
    user_message: message,
    task: session?.task ?? message,
    messages: messages,
    recent_context: __scope_classifier_recent_context(messages, 3),
    workspace_anchor: anchor,
    provider: turn_opts?.provider ?? "",
    model: turn_opts?.model ?? "",
  }
  let outcome = try {
    classifier(payload)
  }
  let verdict = if is_err(outcome) {
    let err = unwrap_err(outcome)
    if error_category(err) == "cancelled" {
      throw err
    }
    __scope_classifier_fail_open(session.session_id, iteration, err)
  } else {
    __scope_classifier_normalize(unwrap(outcome), session.session_id, iteration)
  }
  if verdict != nil {
    agent_emit_event(session.session_id, "scope_classifier_verdict", verdict)
  }
  return verdict
}

fn __scope_classifier_mounted_roots(anchor) {
  let roots = anchor?.additional_roots ?? []
  if type_of(roots) != "list" || len(roots) == 0 {
    return "  (none)"
  }
  var lines = []
  for root in roots {
    lines = lines
      .push(
      "  - " + to_string(root?.path ?? root?.root ?? "")
        + " (mount_mode: "
        + to_string(root?.mount_mode ?? "")
        + ")",
    )
  }
  return join(lines, "\n")
}

fn __scope_classifier_alert_body(verdict, session) {
  let anchor = verdict?.workspace_anchor ?? agent_session_workspace_anchor(session.session_id)
  let primary = to_string(anchor?.primary ?? "(none)")
  return "<scope-alert>\nThe latest user turn appears outside the current workspace anchor. "
    + to_string(verdict?.evidence ?? "")
    + "\n\nCurrent anchor: "
    + primary
    + "\nMounted roots:\n"
    + __scope_classifier_mounted_roots(anchor)
    + "\n\nThree options:\n"
    + "  - add_root: mount the target repo into this session with `agent_session_add_root(session_id, root, {mount_mode})`\n"
    + "  - reanchor: switch the session primary anchor with `agent_session_reanchor(session_id, new_anchor)`\n"
    + "  - fork: spawn a sub-agent against the target repo with `spawn_agent({anchor: new_anchor, ...})`\n\n"
    + "Ask the user which handoff they prefer before doing workspace-mutating work.\n</scope-alert>"
}

fn __scope_classifier_assistant_text(verdict) {
  let evidence = trim(to_string(verdict?.evidence ?? ""))
  let suffix = if evidence == "" {
    ""
  } else {
    " " + evidence
  }
  return "This task appears to be outside the current workspace anchor."
    + suffix
    + " Options: 1) Add Root, 2) Re-anchor, 3) Fork to a new session. Which would you prefer?"
}

fn __scope_classifier_skip_main(verdict) {
  return verdict != nil && verdict?.label == "out_of_scope" && (verdict?.skip_main_turn ?? true)
}

fn __scope_classifier_record_skip_turn(session, verdict, iteration_index) {
  let text = __scope_classifier_assistant_text(verdict)
  agent_session_record_assistant(
    session.session_id,
    {
      text: text,
      visible_text: text,
      provider: "harn",
      model: "scope_classifier",
      input_tokens: 0,
      output_tokens: 0,
      scope_classifier_verdict: verdict,
    },
  )
  agent_session_inject(
    session.session_id,
    transcript_reminder_event(
      {
        body: __scope_classifier_alert_body(verdict, session),
        source: "in_pipeline",
        tags: ["scope_alert", "pre_turn_scope_classifier"],
        dedupe_key: "scope_alert:pre_turn:" + substring(sha256(session.session_id + text), 0, 16),
        ttl_turns: 3,
        fired_at_turn: iteration_index + 1,
      },
    ),
  )
  agent_emit_event(
    session.session_id,
    "iteration_end",
    {
      iteration: iteration_index + 1,
      iteration_info: __iteration_info_payload(
        {
          text: text,
          visible_text: text,
          provider: "harn",
          model: "scope_classifier",
          input_tokens: 0,
          output_tokens: 0,
        },
        0,
        text,
      )
        + {
        dispatch_skipped: true,
        skip_reason: "scope_classifier_out_of_scope",
        scope_classifier_verdict: verdict,
      },
    },
  )
  let _ = __agent_loop_checkpoint(session.session_id, "iteration_end", {iteration: iteration_index + 1})
  return text
}

fn __agent_loop_pop_structural_veto_turn(session_id) {
  let messages = agent_session_messages(session_id)
  if len(messages) == 0 {
    return false
  }
  let last = messages[len(messages) - 1]
  if last?.role != "assistant" {
    return false
  }
  return agent_session_pop_last_assistant(session_id)
}

/**
 * Build the one-shot directive injected as a final system fragment when the
 * loop terminates mid-tool-use. It tells the model the turn/budget is spent,
 * that no more tool calls are possible, and how to wrap up — honoring the
 * configured `done_sentinel` when present.
 */
fn __agent_loop_wrapup_directive(opts) {
  let sentinel = opts?.done_sentinel
  let base = "You have reached your turn/budget limit and cannot call any more tools. "
    + "Provide your final answer now: briefly summarize what you accomplished and what you "
    + "verified. Do not emit any tool call."
  if sentinel == nil || to_string(sentinel) == "" {
    return base
      + " End with a concise `<user_response>...</user_response>` containing the final answer."
  }
  let tag = to_string(sentinel)
  return base
    + " Emit a concise `<user_response>...</user_response>` with the final answer, then end with "
    + "`<done>"
    + tag
    + "</done>` exactly, as instructed by the response protocol."
}

/**
 * True when the loop's terminal status is an exhaustion/cap (not a clean
 * `done`, a suspension, a scope alert, or a hard provider/terminal error).
 * These are the states where the model never produced a clean terminal
 * response and a wrap-up turn is warranted.
 */
fn __agent_loop_wrapup_eligible_status(final_status) {
  return final_status == "budget_exhausted"
    || final_status == "verify_capped"
    || final_status == "verify_exhausted"
    || final_status == "stuck"
}

fn __agent_loop_terminal_callback_continue_allowed(final_status, stop_reason) {
  if final_status == "stuck" {
    return true
  }
  return final_status == "budget_exhausted" && (stop_reason ?? "") == "max_iterations"
}

fn __agent_loop_terminal_callback_extend_by(budget) {
  let by = budget?.extend_by ?? 0
  if type_of(by) == "int" && by > 0 {
    return by
  }
  return 1
}

/**
 * Forced terminal wrap-up turn. When the loop exits on exhaustion/cap while the
 * last turn still called tools, the surfaced final text would otherwise be a
 * dangling tool-call turn with no clean `<user_response>`/sentinel. This runs
 * exactly one extra LLM call with tools DISABLED to elicit the model's final
 * user-facing answer + completion sentinel, records it as the final assistant
 * turn so `agent_session_finalize` surfaces it, and never mutates
 * `final_status`/`stop_reason`. Defensive: any error falls back to existing
 * behavior. Returns true when a wrap-up response was recorded.
 */
fn __agent_loop_final_wrapup(message, session, opts, final_status, stop_reason, iteration) {
  let directive = __agent_loop_wrapup_directive(opts)
  // Disable tools for this turn so the model cannot dispatch: empty tool
  // surface + a system directive forbidding tool calls. Carry the tool format
  // forward so the response protocol still matches what the model was trained
  // on this run.
  let wrapup_opts = opts
    + {
    tools: nil,
    active_skills: nil,
    skill_catalog_prompt: "",
    _progress_tool_system_prompt_nudge: "",
    transcript_projection: opts?.transcript_projection,
  }
  let turn_prompt = __agent_loop_build_turn_prompt(session, wrapup_opts, iteration)
  var turn_system_parts = []
  if trim(turn_prompt.system ?? "") != "" {
    turn_system_parts = turn_system_parts.push(turn_prompt.system)
  }
  turn_system_parts = turn_system_parts.push(directive)
  let turn_system = join(turn_system_parts, "\n\n")
  let llm_opts = __agent_loop_effective_llm_options(wrapup_opts)
    + {
    messages: turn_prompt.messages,
    session_id: session.session_id,
    tool_format: wrapup_opts?.tool_format,
    tools: nil,
    _iteration: iteration + 1,
    _final_wrapup: true,
  }
  let call = __invoke_llm(message, turn_system, llm_opts)
  if !call.ok {
    return false
  }
  let llm_result = call.value
  let raw_text = llm_result?.text ?? ""
  let parsed = agent_parse_tool_calls(raw_text, nil, wrapup_opts?.tool_format)
  let visible_text = __visible_text(parsed, raw_text)
  if to_string(visible_text) == "" {
    return false
  }
  let normalized = llm_result + {text: visible_text, visible_text: visible_text}
  agent_session_record_assistant(session.session_id, normalized)
  let _ = try {
    agent_emit_event(
      session.session_id,
      "final_wrapup",
      {final_status: final_status, stop_reason: stop_reason ?? "", iteration: iteration},
    )
  }
  return true
}

@complexity(allow)
fn __agent_loop_run(message, session, initial_opts) {
  var opts = initial_opts
  var iteration = 0
  var session_finalized = false
  var audit_background_tasks = []
  let run = try {
    opts = agent_mcp_bootstrap_if_needed(session, opts)
    let primary_llm_opts = __agent_loop_effective_llm_options(opts)
    let primary_provider = to_string(primary_llm_opts?.provider ?? "")
    let primary_model = to_string(primary_llm_opts?.model ?? "")
    var escalation_retry_pending = false
    var stop_reason = nil
    var final_status = ""
    var terminal_error = nil
    var verify_attempts = 0
    var done_judge_invocations = 0
    var done_judge_vetoes = 0
    var verify_completion_judge_invocations = 0
    var verify_completion_judge_vetoes = 0
    var step_judge_attempts = 0
    var structural_validator_attempts = 0
    var consecutive_text_only = 0
    var turns_since_progress = 0
    var fallback_index = 0
    var successful_tools_seen = []
    var rejected_tools_seen = []
    var suspend_result = nil
    var stall_state = agent_stall_initial_state()
    var stall_enabled_seen = false
    var stall_prev_dispatch = nil
    let max_verify_attempts = opts?.max_verify_attempts ?? 20
    let budget = opts?.iteration_budget
      ?? {
      mode: "fixed",
      initial: opts?.max_iterations ?? 50,
      max: opts?.max_iterations ?? 50,
      extend_by: 0,
      expose_decisions: false,
      wall_clock_ms: nil,
      total_cost_usd: nil,
      consecutive_failures: nil,
    }
    var current_max = budget.initial
    var extensions_used = 0
    var budget_decisions = []
    var consecutive_failure_count = 0
    var budget_exhausted_emitted = false
    var last_tool_count = 0
    let loop_start_ms = __agent_loop_clock_now_ms()
    let __reserve_cfg = agent_stall_repair_config(opts?.stall_diagnostics)
    var terminal_write_unverified = false
    var terminal_verify_reserve_remaining = if __reserve_cfg.reserved_terminal_verify {
      __reserve_cfg.reserved_terminal_verify_iterations
    } else {
      0
    }
    var in_terminal_verify_reserve = false
    var terminal_callback_continues = 0
    while true {
      while iteration < current_max {
        let boundary_exhaustion = __agent_loop_budget_exhaustion(
          session.session_id,
          budget,
          iteration,
          nil,
          loop_start_ms,
          current_max,
        )
        if boundary_exhaustion.exhausted {
          __agent_loop_emit_budget_exhausted(session.session_id, boundary_exhaustion)
          budget_exhausted_emitted = true
          budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, boundary_exhaustion.kind)
          final_status = "budget_exhausted"
          stop_reason = boundary_exhaustion.kind
          break
        }
        let checkpoint = __agent_loop_suspend_checkpoint(session, iteration)
        if checkpoint != nil {
          __drain_audit_flushes(audit_background_tasks)
          audit_background_tasks = []
          suspend_result = checkpoint
          final_status = "suspended"
          stop_reason = "suspended"
          break
        }
        if agent_budget_pre_call_blocked(session, opts) {
          final_status = "budget_exhausted"
          break
        }
        let iteration_index = iteration
        let iteration_opts = __agent_loop_effective_llm_options(opts)
        agent_emit_event(
          session.session_id,
          "iteration_start",
          {
            iteration: iteration_index + 1,
            provider: iteration_opts?.provider ?? "",
            model: iteration_opts?.model ?? "",
          },
        )
        try {
          __host_drain_file_edits(session.session_id)
        } catch (e) {
          nil
        }
        __agent_loop_checkpoint(session.session_id, "iteration_start", {iteration: iteration_index + 1})
        var turn_opts = agent_skills_match(session, iteration_opts, iteration_index)
        if turn_opts?._skill_activated_this_turn ?? false {
          opts = agent_reset_tool_surface_narrowing(opts)
          turn_opts = agent_reset_tool_surface_narrowing(turn_opts)
        }
        turn_opts = agent_tool_search_inject_if_needed(turn_opts)
        turn_opts = agent_apply_tool_surface_narrowing(turn_opts)
        let turn_llm_opts = __agent_loop_effective_llm_options(turn_opts)
        __agent_loop_checkpoint(session.session_id, "pre_compact", {iteration: iteration_index + 1})
        agent_autocompact_if_needed(session, turn_llm_opts)
        __agent_loop_checkpoint(session.session_id, "post_compact", {iteration: iteration_index + 1})
        let scope_verdict = __run_pre_turn_scope_classifier(
          turn_llm_opts?._pre_turn_scope_classifier ?? opts?._pre_turn_scope_classifier,
          session,
          message,
          turn_llm_opts,
          iteration_index,
        )
        if __scope_classifier_skip_main(scope_verdict) {
          __scope_classifier_record_skip_turn(session, scope_verdict, iteration_index)
          iteration = iteration + 1
          final_status = "scope_alert"
          stop_reason = "out_of_scope"
          break
        }
        let turn_prompt = __agent_loop_build_turn_prompt(session, turn_llm_opts, iteration_index)
        let llm_opts = turn_llm_opts
          + {
          messages: turn_prompt.messages,
          session_id: session.session_id,
          tool_format: turn_llm_opts.tool_format,
          _iteration: iteration_index + 1,
          _system_fragments: turn_prompt.fragments,
        }
        var call = __invoke_llm_with_autocontinue(
          message,
          turn_prompt.system,
          llm_opts,
          turn_opts,
          session.session_id,
          iteration_index,
        )
        if !call.ok && __agent_loop_is_context_overflow(call?.error) {
          call = __agent_loop_recover_context_overflow(
            call,
            message,
            turn_prompt.system,
            llm_opts,
            turn_opts,
            session,
            iteration_index,
          )
        }
        if !call.ok {
          let failed_provider = to_string(turn_llm_opts?.provider ?? "")
          let failed_model = to_string(turn_llm_opts?.model ?? "")
          let was_escalated = (primary_provider != "" || primary_model != "")
            && (failed_provider != primary_provider || failed_model != primary_model)
          let failure_config = __agent_loop_consecutive_failure_config(budget)
          if was_escalated
            && __agent_loop_tracks_failure(call?.error, failure_config)
            && __agent_loop_is_escalation_transport_failure(call?.error) {
            iteration = iteration + 1
            __agent_loop_emit_provider_error(
              session.session_id,
              iteration_index,
              call,
              turn_llm_opts,
              loop_start_ms,
              false,
              "escalation_aborted_provider_transport",
            )
            final_status = "provider_error"
            stop_reason = "escalation_aborted_provider_transport"
            terminal_error = __agent_loop_transport_abort_error(call?.error)
            break
          }
          if __agent_loop_tracks_failure(call?.error, failure_config) {
            iteration = iteration + 1
            consecutive_failure_count = consecutive_failure_count + 1
            let failure_aggregates = __agent_loop_budget_aggregates(session.session_id, nil, loop_start_ms)
            agent_emit_event(
              session.session_id,
              "iteration_end",
              {
                iteration: iteration_index + 1,
                iteration_info: failure_aggregates
                  + {
                  dispatch_skipped: true,
                  skip_reason: "provider_failure",
                  provider_error: call?.error ?? {},
                  consecutive_failures: consecutive_failure_count,
                },
              },
            )
            if consecutive_failure_count >= failure_config.max {
              let paused_for_ms = failure_config?.paused_for_ms ?? 0
              agent_emit_event(
                session.session_id,
                "budget_circuit_breaker",
                {
                  kind: "consecutive_failures",
                  consecutive_count: consecutive_failure_count,
                  paused_for_ms: paused_for_ms,
                },
              )
              if paused_for_ms > 0 {
                __agent_loop_clock_sleep_ms(paused_for_ms)
              }
              let exhaustion = __agent_loop_budget_aggregates(session.session_id, nil, loop_start_ms)
                + {exhausted: true, kind: "consecutive_failures", iteration: iteration, max_iterations: current_max}
              __agent_loop_emit_budget_exhausted(session.session_id, exhaustion)
              budget_exhausted_emitted = true
              budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, exhaustion.kind)
              final_status = "budget_exhausted"
              stop_reason = "circuit_breaker"
              terminal_error = call?.error
              break
            }
            continue
          }
          let can_retry_primary = was_escalated && !escalation_retry_pending
          __agent_loop_emit_provider_error(
            session.session_id,
            iteration_index,
            call,
            turn_llm_opts,
            loop_start_ms,
            can_retry_primary,
            "",
          )
          if can_retry_primary {
            opts = opts
              + {
              provider: primary_provider,
              model: primary_model,
              llm_options: (opts?.llm_options ?? {})
                + {provider: primary_provider, model: primary_model},
            }
            escalation_retry_pending = true
            iteration = iteration + 1
            continue
          }
          final_status = call.status
          stop_reason = call?.stop_reason ?? call.status
          terminal_error = call?.error
          break
        }
        escalation_retry_pending = false
        let llm_result = call.value
        consecutive_failure_count = 0
        iteration = iteration + 1
        let raw_text = llm_result?.text ?? ""
        let parsed = agent_parse_tool_calls(raw_text, turn_opts?.tools, turn_opts?.tool_format)
        let visible_text = __visible_text(parsed, raw_text)
        let normalized = llm_result
          + {text: visible_text, visible_text: visible_text, _agent_tool_format: turn_opts?.tool_format ?? ""}
        let fallback_outcome = __detect_native_fallback(
          llm_result,
          parsed,
          turn_opts,
          fallback_index,
          session.session_id,
          iteration_index,
        )
        fallback_index = fallback_outcome.fallback_index
        let tool_calls = if fallback_outcome.triggered {
          fallback_outcome.calls ?? []
        } else {
          __resolve_tool_calls(llm_result, parsed)
        }
        let recorded_assistant = if fallback_outcome.triggered && fallback_outcome.accepted {
          normalized + {tool_calls: tool_calls, native_tool_calls: tool_calls}
        } else {
          normalized
        }
        agent_session_record_assistant(session.session_id, recorded_assistant)
        let had_parse_errors_base = __maybe_inject_parse_error_feedback(session.session_id, parsed, tool_calls, turn_opts)
        let had_blank_name_drop = if fallback_outcome.triggered {
          false
        } else {
          __maybe_inject_blank_name_feedback(
            session.session_id,
            llm_result,
            parsed,
            len(tool_calls),
            turn_opts,
          )
        }
        let had_parse_errors = had_parse_errors_base || had_blank_name_drop
        let await_call = __agent_await_resumption_call(tool_calls)
        if await_call != nil {
          let await_result = try {
            __agent_loop_await_resumption(session, iteration, await_call, opts)
          }
          if is_err(await_result) {
            let await_error = unwrap_err(await_result)
            agent_session_inject_feedback(
              session.session_id,
              "agent_await_resumption",
              __agent_loop_invalid_await_resumption_feedback(await_error),
            )
            let await_totals = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
            agent_emit_event(
              session.session_id,
              "iteration_end",
              {
                iteration: iteration_index + 1,
                iteration_info: __agent_loop_iteration_info(
                  session.session_id,
                  llm_result,
                  len(tool_calls),
                  visible_text,
                  await_totals,
                  loop_start_ms,
                )
                  + {dispatch_skipped: true, skip_reason: "invalid_agent_await_resumption"},
              },
            )
            let await_exhaustion = __agent_loop_budget_exhaustion(
              session.session_id,
              budget,
              iteration,
              await_totals,
              loop_start_ms,
              current_max,
            )
            if await_exhaustion.exhausted {
              __agent_loop_emit_budget_exhausted(session.session_id, await_exhaustion)
              budget_exhausted_emitted = true
              budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, await_exhaustion.kind)
              final_status = "budget_exhausted"
              stop_reason = await_exhaustion.kind
              break
            }
            continue
          }
          suspend_result = unwrap(await_result)
          let _totals = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
          agent_emit_event(
            session.session_id,
            "iteration_end",
            {
              iteration: iteration_index + 1,
              iteration_info: __agent_loop_iteration_info(
                session.session_id,
                llm_result,
                len(tool_calls),
                visible_text,
                _totals,
                loop_start_ms,
              ),
            },
          )
          final_status = "suspended"
          stop_reason = "suspended"
          break
        }
        let stall_judge_due = agent_stall_done_judge_due(turn_opts, done_judge_invocations, iteration_index + 1)
        let stall_prev_turn_made_edit = __turn_made_edit(stall_prev_dispatch, opts?.tools)
        let stall_observation = agent_stall_observe_tool_calls(
          session.session_id,
          tool_calls,
          iteration_index + 1,
          turn_opts?.stall_diagnostics,
          stall_state,
          stall_judge_due,
          stall_prev_dispatch,
          visible_text,
          had_parse_errors,
          stall_prev_turn_made_edit,
          llm_result?.stop_reason ?? "",
        )
        stall_state = stall_observation.state
        stall_enabled_seen = stall_enabled_seen || stall_observation.enabled
        if stall_state.last_diagnostic_class == "pass" {
          terminal_write_unverified = false
        }
        let stall_warning = stall_observation.warning
        let structural_verdict = __run_structural_validator(
          opts?._structural_validator,
          session.session_id,
          normalized + {raw_text: raw_text},
          tool_calls,
          parsed,
          llm_opts,
          turn_opts,
          successful_tools_seen,
          rejected_tools_seen,
          structural_validator_attempts,
        )
        if structural_verdict.vetoed {
          let on_failure = structural_verdict?.on_failure ?? "regenerate_with_feedback"
          if on_failure == "raise" {
            throw structural_verdict?.diagnostic
              ?? "structural validator rejected assistant turn"
          }
          structural_validator_attempts = structural_validator_attempts + 1
          __agent_loop_pop_structural_veto_turn(session.session_id)
          let feedback = to_string(structural_verdict?.feedback ?? "")
          if feedback != "" {
            agent_session_inject_feedback(session.session_id, "structural_validator", feedback)
          }
          let structural_totals = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
          agent_emit_event(
            session.session_id,
            "iteration_end",
            {
              iteration: iteration_index + 1,
              iteration_info: __agent_loop_iteration_info(
                session.session_id,
                llm_result,
                0,
                visible_text,
                structural_totals,
                loop_start_ms,
              )
                + {
                dispatch_skipped: true,
                skip_reason: "structural_validator_revise",
                structural_validator_attempts: structural_validator_attempts,
                structural_validator_rule: structural_verdict?.rule ?? "",
              },
            },
          )
          let structural_exhaustion = __agent_loop_budget_exhaustion(
            session.session_id,
            budget,
            iteration,
            structural_totals,
            loop_start_ms,
            current_max,
          )
          if structural_exhaustion.exhausted {
            __agent_loop_emit_budget_exhausted(session.session_id, structural_exhaustion)
            budget_exhausted_emitted = true
            budget_decisions = __agent_loop_record_budget_stop(
              budget_decisions,
              iteration,
              current_max,
              structural_exhaustion.kind,
            )
            final_status = "budget_exhausted"
            stop_reason = structural_exhaustion.kind
            break
          }
          continue
        } else if !(structural_verdict?.skipped ?? false) {
          structural_validator_attempts = 0
        }
        if stall_judge_due && stall_warning != nil {
          let stall_verify_opts = turn_opts
            + {
            _done_judge_due: true,
            _done_judge_trigger: "stalled",
            _done_judge_invocations: done_judge_invocations,
          }
          let stall_verdict = agent_verify_or_continue(session, stall_verify_opts, "stalled", visible_text, iteration_index + 1)
          if stall_verdict?.done_judge_invoked ?? false {
            done_judge_invocations = done_judge_invocations + 1
          }
          if stall_verdict?.done_judge_cap_reached ?? false {
            final_status = "verify_capped"
            stop_reason = "done_judge_cap_reached"
            break
          }
          if stall_verdict.vetoed {
            if stall_verdict?.done_judge_invoked ?? false {
              done_judge_vetoes = done_judge_vetoes + 1
            }
            if stall_observation.feedback_deferred {
              stall_state = agent_stall_inject_feedback(
                session.session_id,
                stall_warning,
                stall_observation.config,
                stall_state,
              )
            }
          } else {
            let stall_totals = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
            let tool_count = len(tool_calls)
            agent_emit_event(
              session.session_id,
              "iteration_end",
              {
                iteration: iteration_index + 1,
                iteration_info: __agent_loop_iteration_info(
                  session.session_id,
                  llm_result,
                  tool_count,
                  visible_text,
                  stall_totals,
                  loop_start_ms,
                ),
              },
            )
            let stall_exhaustion = __agent_loop_budget_exhaustion(
              session.session_id,
              budget,
              iteration,
              stall_totals,
              loop_start_ms,
              current_max,
            )
            if stall_exhaustion.exhausted {
              __agent_loop_emit_budget_exhausted(session.session_id, stall_exhaustion)
              budget_exhausted_emitted = true
              budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, stall_exhaustion.kind)
              final_status = "budget_exhausted"
              stop_reason = stall_exhaustion.kind
              break
            }
            let stalled_done_checkpoint = __agent_loop_checkpoint(session.session_id, "iteration_end", {iteration: iteration_index + 1})
            if stalled_done_checkpoint.delivered > 0 {
              continue
            }
            final_status = "done"
            stop_reason = "stalled_done_judge"
            break
          }
        }
        if stall_observation.hard_stop {
          agent_emit_event(
            session.session_id,
            "loop_control_decision",
            {
              iteration: iteration_index + 1,
              action: "stop",
              old_limit: current_max,
              new_limit: current_max,
              reason: "thrash_hard_stop",
              status: "stuck",
            },
          )
          final_status = "stuck"
          stop_reason = "thrash_hard_stop"
          break
        }
        if turn_opts?.step_judge != nil {
          let remaining_iterations = current_max - iteration_index
          let step_verdict = agent_step_judge(
            session,
            llm_result,
            turn_opts,
            iteration_index + 1,
            stall_warning,
            step_judge_attempts,
            remaining_iterations,
          )
          if step_verdict.vetoed {
            step_judge_attempts = step_judge_attempts + 1
            let on_veto = step_verdict?.on_veto ?? "replace"
            if on_veto == "replace" {
              agent_session_pop_last_assistant(session.session_id)
            }
            let critique = step_verdict?.feedback ?? step_verdict?.critique ?? ""
            if critique != "" {
              agent_session_inject_feedback(session.session_id, "step_judge", critique)
            }
            let step_totals = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
            agent_emit_event(
              session.session_id,
              "iteration_end",
              {
                iteration: iteration_index + 1,
                iteration_info: __agent_loop_iteration_info(
                  session.session_id,
                  llm_result,
                  0,
                  visible_text,
                  step_totals,
                  loop_start_ms,
                )
                  + {
                  dispatch_skipped: true,
                  skip_reason: "step_judge_revise",
                  on_veto: on_veto,
                  step_judge_attempts: step_judge_attempts,
                },
              },
            )
            let step_exhaustion = __agent_loop_budget_exhaustion(
              session.session_id,
              budget,
              iteration,
              step_totals,
              loop_start_ms,
              current_max,
            )
            if step_exhaustion.exhausted {
              __agent_loop_emit_budget_exhausted(session.session_id, step_exhaustion)
              budget_exhausted_emitted = true
              budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, step_exhaustion.kind)
              final_status = "budget_exhausted"
              stop_reason = step_exhaustion.kind
              break
            }
            continue
          } else if !(step_verdict?.skipped ?? false) {
            step_judge_attempts = 0
          }
        }
        let pre_dispatch_checkpoint = __agent_loop_checkpoint(session.session_id, "pre_tool_dispatch", {iteration: iteration_index + 1})
        if pre_dispatch_checkpoint.dispatch_skipped {
          let tool_count_skipped = len(tool_calls)
          let totals_skipped = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
          agent_emit_event(
            session.session_id,
            "iteration_end",
            {
              iteration: iteration_index + 1,
              iteration_info: __agent_loop_iteration_info(
                session.session_id,
                llm_result,
                tool_count_skipped,
                visible_text,
                totals_skipped,
                loop_start_ms,
              )
                + {dispatch_skipped: true, skip_reason: "interrupt_immediate"},
            },
          )
          let skipped_exhaustion = __agent_loop_budget_exhaustion(
            session.session_id,
            budget,
            iteration,
            totals_skipped,
            loop_start_ms,
            current_max,
          )
          if skipped_exhaustion.exhausted {
            __agent_loop_emit_budget_exhausted(session.session_id, skipped_exhaustion)
            budget_exhausted_emitted = true
            budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, skipped_exhaustion.kind)
            final_status = "budget_exhausted"
            stop_reason = skipped_exhaustion.kind
            break
          }
          if agent_budget_post_call_blocked(totals_skipped, turn_opts) {
            final_status = "budget_exhausted"
            break
          }
          consecutive_text_only = __next_text_only_count(0, consecutive_text_only)
          turns_since_progress = __next_progress_count(false, turns_since_progress)
          last_tool_count = 0
          continue
        }
        let dispatched = __dispatch_tool_calls(
          session.session_id,
          tool_calls,
          turn_opts
            + {
            _iteration: iteration_index + 1,
            _tool_caller: opts?._tool_caller,
            _stop_reason: llm_result?.stop_reason ?? "",
          },
        )
        let dispatch = dispatched.dispatch
        stall_prev_dispatch = dispatch
        audit_background_tasks = __spawn_audit_flushes(audit_background_tasks, dispatched.audit_flushes)
        opts = __sync_tool_search_state(opts, dispatched.turn_opts)
        successful_tools_seen = __merge_tool_names(successful_tools_seen, __tool_names_by_status(dispatch, true))
        rejected_tools_seen = __merge_tool_names(rejected_tools_seen, __tool_names_by_status(dispatch, false))
        let totals = agent_session_record_usage(session.session_id, llm_result, turn_llm_opts, iteration_index + 1)
        let tool_count = len(tool_calls)
        agent_emit_event(
          session.session_id,
          "iteration_end",
          {
            iteration: iteration_index + 1,
            iteration_info: __agent_loop_iteration_info(
              session.session_id,
              llm_result,
              tool_count,
              visible_text,
              totals,
              loop_start_ms,
            ),
          },
        )
        let post_dispatch_checkpoint = __agent_loop_checkpoint(session.session_id, "post_tool_dispatch", {iteration: iteration_index + 1})
        let bridge_step_delivered = post_dispatch_checkpoint.delivered
        let exhaustion = __agent_loop_budget_exhaustion(
          session.session_id,
          budget,
          iteration,
          totals,
          loop_start_ms,
          current_max,
        )
        if exhaustion.exhausted {
          __agent_loop_emit_budget_exhausted(session.session_id, exhaustion)
          budget_exhausted_emitted = true
          budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, exhaustion.kind)
          final_status = "budget_exhausted"
          stop_reason = exhaustion.kind
          break
        }
        if agent_budget_post_call_blocked(totals, turn_opts) {
          final_status = "budget_exhausted"
          break
        }
        consecutive_text_only = __next_text_only_count(tool_count, consecutive_text_only)
        last_tool_count = tool_count
        let turn_successful = __tool_names_by_status(dispatch, true)
        let turn_rejected = __tool_names_by_status(dispatch, false)
        let made_progress = len(turn_successful) > 0
        turns_since_progress = __next_progress_count(made_progress, turns_since_progress)
        let turn_max_nudges = turn_opts?.max_nudges ?? 8
        let turn_loop_until_done = turn_opts?.loop_until_done ?? false
        let text_only_nudge_budget_exceeded = turn_loop_until_done
          && tool_count == 0
          && consecutive_text_only > turn_max_nudges
        let missing_required_for_loop = agent_required_tools_missing_from_session(opts, successful_tools_seen)
        let cadence_loop_state = agent_loop_snapshot_state(
          {
            iteration: iteration,
            current_limit: current_max,
            budget_max: budget.max,
            extensions_used: extensions_used,
            tool_count: tool_count,
            turn_successful: turn_successful,
            turn_rejected: turn_rejected,
            visible_text: visible_text,
            turn_native_fallback_used: fallback_outcome.triggered && fallback_outcome.accepted,
            session_successful: successful_tools_seen,
            session_rejected: rejected_tools_seen,
            missing_required_tools: missing_required_for_loop,
            completion_proposed: false,
            verdict: nil,
          }
            + __agent_loop_state_budget_fields(
            session.session_id,
            totals,
            loop_start_ms,
            budget,
            consecutive_failure_count,
          ),
        )
        let post_turn_opts = turn_opts
          + {
          _session_successful_tools: successful_tools_seen,
          _session_rejected_tools: rejected_tools_seen,
          _consecutive_text_only: consecutive_text_only,
          _done_judge_invocations: done_judge_invocations,
          _done_judge_loop_state: cadence_loop_state,
          _turn_tool_call_feedback: (fallback_outcome.triggered && !fallback_outcome.accepted)
            || had_parse_errors,
        }
        let outcome = __agent_loop_with_llm_render_context(
          turn_llm_opts,
          { _effective -> agent_compute_post_turn(
            session,
            normalized + {raw_text: raw_text, parsed_done_marker: parsed?.done_marker ?? ""},
            dispatch,
            post_turn_opts,
            iteration_index,
          ) },
        )
        opts = __apply_post_turn_options(opts, outcome)
        let repair_cfg = agent_stall_repair_config(turn_opts?.stall_diagnostics)
        let reverify_mandated = __agent_loop_post_edit_reverify_mandated(repair_cfg, turn_opts, stall_state, dispatch, opts?.tools)
        if __turn_made_edit(dispatch, opts?.tools) {
          terminal_write_unverified = true
        }
        var should_continue = outcome.kind == "continue" || bridge_step_delivered > 0
        var verdict_record = nil
        var post_edit_reverify_confirmed = false
        if should_continue && reverify_mandated && verify_attempts < max_verify_attempts {
          let reverify_opts = turn_opts
            + {
            _done_judge_due: false,
            _done_judge_invocations: done_judge_invocations,
            _verify_completion_judge_invocations: verify_completion_judge_invocations,
          }
          let reverdict = agent_verify_or_continue(
            session,
            reverify_opts,
            "post_edit_reverify",
            llm_result.text,
            iteration_index,
          )
          verdict_record = reverdict
          if reverdict?.verify_completion_judge_invoked ?? false {
            verify_completion_judge_invocations = verify_completion_judge_invocations + 1
          }
          if reverdict.vetoed {
            verify_attempts = verify_attempts + 1
            if reverdict?.verify_completion_judge_invoked ?? false {
              verify_completion_judge_vetoes = verify_completion_judge_vetoes + 1
            }
          } else {
            stall_state = stall_state + {reverify_owed: false}
            post_edit_reverify_confirmed = true
            terminal_write_unverified = false
          }
        }
        if should_continue {
          if opts?.daemon && len(tool_calls) == 0 {
            agent_daemon_step(session, opts, iteration)
          }
        } else {
          if outcome.needs_verify || reverify_mandated {
            if verify_attempts >= max_verify_attempts {
              final_status = "verify_exhausted"
              stop_reason = outcome.stop_reason
              break
            }
            let verify_opts = turn_opts
              + {
              _done_judge_due: outcome?.done_judge_due ?? true,
              _done_judge_invocations: done_judge_invocations,
              _verify_completion_judge_invocations: verify_completion_judge_invocations,
            }
            let verdict = agent_verify_or_continue(
              session,
              verify_opts,
              outcome.stop_reason,
              llm_result.text,
              iteration_index,
            )
            verdict_record = verdict
            if verdict?.done_judge_invoked ?? false {
              done_judge_invocations = done_judge_invocations + 1
            }
            if verdict?.verify_completion_judge_invoked ?? false {
              verify_completion_judge_invocations = verify_completion_judge_invocations + 1
            }
            if verdict.vetoed {
              verify_attempts = verify_attempts + 1
              should_continue = true
              if verdict?.done_judge_invoked ?? false {
                done_judge_vetoes = done_judge_vetoes + 1
              }
              if verdict?.verify_completion_judge_invoked ?? false {
                verify_completion_judge_vetoes = verify_completion_judge_vetoes + 1
              }
            } else if verdict?.verify_completion_judge_cap_reached ?? false {
              final_status = "verify_capped"
              stop_reason = "completion_judge_cap_reached"
              break
            } else if verdict?.done_judge_cap_reached ?? false {
              final_status = "verify_capped"
              stop_reason = "done_judge_cap_reached"
              break
            } else {
              stall_state = stall_state + {reverify_owed: false}
              terminal_write_unverified = false
              if reverify_mandated {
                post_edit_reverify_confirmed = true
              }
            }
          }
          let missing_now = agent_required_tools_missing_from_session(opts, successful_tools_seen)
          if !should_continue && len(missing_now) > 0 {
            agent_required_tools_inject_feedback(session.session_id, missing_now)
            should_continue = true
          }
          if !should_continue {
            stop_reason = outcome.stop_reason
            break
          }
        }
        if should_continue && post_edit_reverify_confirmed && iteration >= current_max {
          let missing_after_reverify = agent_required_tools_missing_from_session(opts, successful_tools_seen)
          if len(missing_after_reverify) == 0 {
            final_status = "done"
            stop_reason = "post_edit_reverify"
            break
          }
        }
        if should_continue {
          if text_only_nudge_budget_exceeded {
            final_status = "stuck"
            stop_reason = "max_nudges"
            break
          }
          agent_scratchpad_reorganize_if_due(
            session,
            opts,
            iteration_index,
            {
              reason: "iteration_end",
              outcome_kind: outcome.kind,
              bridge_step_delivered: bridge_step_delivered,
              tool_count: tool_count,
            },
          )
          if turns_since_progress >= 2
            && turns_since_progress <= turn_max_nudges
            && !(outcome?.nudged_this_turn ?? false)
            && len(turn_rejected) == 0 {
            let has_tools = len(opts?.tools?.tools ?? []) > 0
            let tool_mode = agent_tool_call_paradigm(turn_opts).kind
            let made_tool_calls = tool_count > 0
            agent_session_inject_feedback(
              session.session_id,
              "no_progress_streak",
              __progress_nudge_text(turns_since_progress, has_tools, tool_mode, made_tool_calls),
            )
            agent_emit_event(
              session.session_id,
              "no_progress_streak_nudge",
              {
                iteration: iteration_index,
                turns_since_progress: turns_since_progress,
                has_tools: has_tools,
                made_tool_calls: made_tool_calls,
              },
            )
          }
        }
        let loop_no_net_progress = agent_stall_no_net_progress(turn_opts?.stall_diagnostics, stall_state)
        let loop_state = agent_loop_snapshot_state(
          {
            iteration: iteration,
            current_limit: current_max,
            budget_max: budget.max,
            extensions_used: extensions_used,
            tool_count: tool_count,
            turn_successful: turn_successful,
            turn_rejected: turn_rejected,
            visible_text: visible_text,
            turn_native_fallback_used: fallback_outcome.triggered && fallback_outcome.accepted,
            session_successful: successful_tools_seen,
            session_rejected: rejected_tools_seen,
            missing_required_tools: missing_required_for_loop,
            completion_proposed: outcome.kind == "break",
            verdict: verdict_record,
            progress_no_net_advance: loop_no_net_progress,
          }
            + __agent_loop_state_budget_fields(
            session.session_id,
            totals,
            loop_start_ms,
            budget,
            consecutive_failure_count,
          ),
        )
        let command = agent_loop_control_invoke(opts, budget, loop_state)
        let applied = agent_loop_apply_command(
          {
            command: command,
            session_id: session.session_id,
            iteration: iteration,
            current_max: current_max,
            extensions_used: extensions_used,
            decisions: budget_decisions,
            budget: budget,
          },
        )
        current_max = applied.current_max
        extensions_used = applied.extensions_used
        budget_decisions = applied.decisions
        if applied.stop {
          final_status = applied.final_status
          stop_reason = applied.stop_reason
          break
        }
      }
      if final_status == "" && iteration >= current_max && stop_reason == nil {
        final_status = "budget_exhausted"
        stop_reason = stop_reason ?? "max_iterations"
      }
      if __agent_loop_should_spend_reserve(
        __reserve_cfg,
        opts,
        final_status,
        terminal_write_unverified,
        verify_attempts < max_verify_attempts,
      ) {
        in_terminal_verify_reserve = true
        let terminal_verdict = agent_verify_or_continue(
          session,
          opts
            + {
            _done_judge_due: false,
            _done_judge_invocations: done_judge_invocations,
            _verify_completion_judge_invocations: verify_completion_judge_invocations,
          },
          "reserved_terminal_verify",
          "",
          iteration,
        )
        if terminal_verdict?.verify_completion_judge_invoked ?? false {
          verify_completion_judge_invocations = verify_completion_judge_invocations + 1
        }
        if !terminal_verdict.vetoed {
          terminal_write_unverified = false
          stall_state = stall_state + {reverify_owed: false}
          final_status = "done"
          stop_reason = "reserved_terminal_verify"
          agent_emit_event(
            session.session_id,
            "reserved_terminal_verify",
            {phase: "verify_passed", iteration: iteration},
          )
          break
        }
        verify_attempts = verify_attempts + 1
        if terminal_verdict?.verify_completion_judge_invoked ?? false {
          verify_completion_judge_vetoes = verify_completion_judge_vetoes + 1
        }
        agent_emit_event(
          session.session_id,
          "reserved_terminal_verify",
          {phase: "verify_failed", iteration: iteration, terminal_status: final_status},
        )
        if terminal_verify_reserve_remaining > 0 && verify_attempts < max_verify_attempts {
          terminal_verify_reserve_remaining = terminal_verify_reserve_remaining - 1
          agent_emit_event(
            session.session_id,
            "reserved_terminal_verify",
            {
              phase: "grant",
              reserve_remaining: terminal_verify_reserve_remaining,
              iteration: iteration,
              prior_status: final_status,
            },
          )
          current_max = current_max + 1
          final_status = ""
          stop_reason = nil
          budget_exhausted_emitted = false
          continue
        }
        stop_reason = "reserved_terminal_verify_failed"
      }
      if final_status != ""
        && suspend_result == nil
        && terminal_error == nil
        && terminal_callback_continues == 0
        && __agent_loop_terminal_callback_continue_allowed(final_status, stop_reason) {
        let terminal_outcome = agent_compute_terminal_callback(
          session,
          opts,
          {
            iteration: iteration,
            final_status: final_status,
            stop_reason: stop_reason ?? "",
            max_iterations: current_max,
            iteration_budget: budget,
          },
        )
        if terminal_outcome.kind == "continue" {
          terminal_callback_continues = terminal_callback_continues + 1
          opts = __apply_post_turn_options(opts, terminal_outcome)
          let old_limit = current_max
          let extra = __agent_loop_terminal_callback_extend_by(budget)
          current_max = max(current_max, iteration + extra)
          budget_decisions = __agent_loop_record_terminal_callback_continue(budget_decisions, iteration, old_limit, current_max)
          agent_emit_event(
            session.session_id,
            "loop_control_decision",
            {
              iteration: iteration,
              action: "extend",
              old_limit: old_limit,
              new_limit: current_max,
              reason: "terminal_callback_continue",
              status: "",
            },
          )
          final_status = ""
          stop_reason = nil
          budget_exhausted_emitted = false
          continue
        }
      }
      break
    }
    if final_status == "budget_exhausted" && !budget_exhausted_emitted {
      let terminal_exhaustion = __agent_loop_budget_aggregates(session.session_id, nil, loop_start_ms)
        + {
        exhausted: true,
        kind: stop_reason ?? "max_iterations",
        iteration: iteration,
        max_iterations: current_max,
      }
      __agent_loop_emit_budget_exhausted(session.session_id, terminal_exhaustion)
      budget_exhausted_emitted = true
      budget_decisions = __agent_loop_record_budget_stop(budget_decisions, iteration, current_max, terminal_exhaustion.kind)
    }
    let wrapup_enabled = opts?.final_wrapup ?? true
    if wrapup_enabled
      && suspend_result == nil
      && terminal_error == nil
      && last_tool_count > 0
      && __agent_loop_wrapup_eligible_status(final_status) {
      let _ = try {
        __agent_loop_final_wrapup(message, session, opts, final_status, stop_reason, iteration)
      }
    }
    if opts?.daemon && final_status != "" {
      agent_daemon_snapshot(session, opts, final_status, iteration)
    }
    try {
      __host_drain_file_edits(session.session_id)
    } catch (e) {
      nil
    }
    __agent_loop_checkpoint(session.session_id, "loop_exit", {iteration: iteration})
    __drain_audit_flushes(audit_background_tasks)
    audit_background_tasks = []
    let result = agent_session_finalize(
      session.session_id,
      {
        final_status: final_status,
        stop_reason: stop_reason ?? "",
        iterations: iteration,
        error: terminal_error,
      },
    )
    session_finalized = true
    let terminated_ok = terminal_error == nil
      && suspend_result == nil
      && (final_status == "" || final_status == "done")
    let final_stall_state = if terminated_ok {
      agent_stall_clear_current_failure(stall_state)
    } else {
      stall_state
    }
    let result_with_stalls = agent_stall_apply_result(result, stall_enabled_seen, final_stall_state)
    let enforced = if suspend_result != nil {
      result_with_stalls
    } else {
      agent_required_tools_enforce(result_with_stalls, opts)
    }
    var final_result = enforced
    if opts?.verify_completion_judge != nil {
      final_result = final_result
        + {
        completion_judge: {
          invocations: verify_completion_judge_invocations,
          vetoes: verify_completion_judge_vetoes,
          max_invocations: agent_verify_completion_judge_cap(opts?.verify_completion_judge),
          cap_reached: final_status == "verify_capped",
        },
      }
    }
    if opts?.done_judge != nil {
      final_result = final_result
        + {
        done_judge: {
          invocations: done_judge_invocations,
          vetoes: done_judge_vetoes,
          max_invocations: agent_done_judge_cap(opts?.done_judge),
          cap_reached: final_status == "verify_capped" && stop_reason == "done_judge_cap_reached",
        },
      }
    }
    if budget.expose_decisions {
      final_result = final_result
        + {
        adaptive_budget: {
          mode: budget.mode,
          initial: budget.initial,
          max: budget.max,
          final_limit: current_max,
          extensions_used: extensions_used,
          decisions: budget_decisions,
        },
      }
    }
    if suspend_result != nil {
      final_result = final_result + suspend_result
    }
    final_result
  }
  if is_err(run) {
    __drain_audit_flushes(audit_background_tasks)
    if !session_finalized {
      __agent_loop_finalize_failed(session, iteration)
    }
    throw unwrap_err(run)
  }
  return unwrap(run)
}

// Reserved terminal-verify guard (default OFF). `terminal_write_unverified`
// tracks whether the transcript has a source write that no passing
// verification has cleared; `terminal_verify_reserve_remaining` is the
// held-back iteration allowance the main loop cannot consume — it is only
// granted (as a bounded extension of `current_max`) when the loop would
// otherwise terminate on a budget/stuck boundary with an unverified red
// edit, so the run runs a final verify(+repair) instead of stopping blind.
// A genuine PASSING verification this turn (the diagnostic fold cleared the
// current-failure model to "pass" on a non-edit turn) is real evidence the
// workspace is green, so the reserved guard's unverified-write debt is
// satisfied. Edit turns that report "pass" are the edit's own success, not
// a verification, and the fold keeps the failure model armed there.
// Reserved terminal-verify bookkeeping: any successful source edit leaves
// the transcript with an unverified write until a verification passes.
// Broader twin of `reverify_owed` (which arms only after a PRIOR failure);
// the reserved guard fires on ANY unverified edit so a model that writes
// source then exhausts budget without ever testing is still caught. Cleared
// at every verify-pass point below (mirrors the `reverify_owed: false`
// clears) and on a passing diagnostic fold above.
// The iteration-count boundary (`while iteration < current_max` falling
// through) terminates without setting `final_status`; resolve it to the
// real budget terminal here so the reserved-verify guard below sees an
// accurate status before deciding whether to spend the reserve.
// Reserved terminal-verify guard (default OFF). The inner loop has stopped.
// If the run is terminating on a budget/stuck boundary while a source write
// is still unverified, run the declared verifier ONCE here (verify-first).
// A green build completes the run `done`; a red build surfaces the actual
// verifier failure (never a silent `budget_exhausted`) and, if reserve
// iterations remain, grants exactly one bounded repair turn before
// re-verifying on the next terminal pass. The reserve lives OUTSIDE the
// main `current_max`, so normal iterations cannot consume it, and it is
// hard-capped so it can never loop forever. This is the budget-exit
// counterpart to #3629's loop-cap done-conversion (which fires only when an
// in-loop reverify ALREADY confirmed green). When OFF the reserve is 0 and
// this whole block is a no-op — byte-identical to today.
// Green: the unverified write is now verified; complete the run.
// Red: the verify RAN and its failure is surfaced (feedback already
// injected by agent_verify_or_continue), so the run can never end blind.
// Spend one reserve iteration on a bounded repair turn; the model reads
// the injected verify failure and edits, then we re-verify next pass.
// Reserve exhausted: keep the terminal status but stamp that the verify
// ran and failed, so the terminal reflects the real (red) build state.
// Whether THIS turn actually issued tool calls. When it did, the
// "narration loop" framing is false (the model is acting, not narrating)
// so the nudge steers toward a DIFFERENT change instead.
// Capture the PRIMARY provider/model once, before any post-turn escalation
// can rebind `opts`. A failed escalated turn falls back to this for one
// retry instead of unwinding the whole loop into a fake success.
// Non-tracked provider failure. A fast pre-dispatch escalation failure must not
// unwind the loop while ACP returns the prior cheap-model text as a fake
// success. Two fixes, always correct:
//   1. Always emit an observable provider_error `iteration_end` event.
//   2. If this was an ESCALATED turn (current provider/model differs
//      from the captured primary) and we have not already retried the
//      primary for this failure, degrade to the PRIMARY provider for
//      one more turn instead of unwinding into a fake success.
// Rebind opts back to the primary provider/model and retry one turn.
// Recoverable: emergency-compact the transcript and retry the turn
// before treating overflow as a failure. The agent must keep working
// on a large repo, not die when the prompt outgrows the window.
// A SUCCESSFUL terminal hand-back (clean done / a passing verify_completion)
// must not report a stale current_failure: a run can complete without ever
// flowing a passing verification result through the fold. Clear the
// current-failure model on the successful-termination signal (mirrors the
// host's `final_status.is_empty() || == "done"` success predicate) before
// projecting it; a stuck / exhausted / suspended run keeps the failure.
/**
 * agent_loop.
 *
 * @effects: [host, agent]
 * @errors: []
 * @api_stability: experimental
 */
pub fn agent_loop(message, system_prompt = nil, options = nil) {
  var opts = agent_loop_options(agent_progress_apply_options(options))
  opts = opts + {tools: agent_lifecycle_tools(opts?.tools, opts)}
  if system_prompt != nil && system_prompt != "" {
    opts = opts + {system: system_prompt}
  }
  let session = agent_session_init(message, system_prompt, opts)
  if session?.done {
    return session.result
  }
  agent_scratchpad_init(session, opts)
  if opts?._tool_format_override != nil {
    agent_emit_event(session.session_id, "tool_format_override", opts._tool_format_override)
  }
  if opts?._tool_format_capability_gap != nil {
    agent_emit_event(session.session_id, "capability_gap", opts._tool_format_capability_gap)
  }
  defer {
    try {
      __host_mcp_disconnect(session.session_id)
    } catch (e) {
    }
  }
  opts = __agent_loop_fire_resume_continuity(session, opts)
  return __agent_loop_run(message, session, opts)
}