import { type Span, SpanStatusCode } from '@opentelemetry/api'
import {
	extractFromAssistantMessage,
	extractFromUserMessage,
} from '../../../compaction/extractor.js'
import { AUTO_CONTINUATION_USER_MESSAGE } from '../../../constants/continuation.js'
import {
	DEFAULT_STRUCTURED_OUTPUT_RETRIES,
	STRUCTURED_OUTPUT_REPROMPT,
} from '../../../constants/tools/index.js'
import { renderSkillsSection } from '../../../persona/assembler.js'
import { resolveProviderCapabilities } from '../../../provider/capabilities.js'
import { collectChatCompletion } from '../../../provider/collect-chat-completion.js'
import { renderToolSchema } from '../../../registry/tool/schema.js'
import { formatCompletionNotification } from '../../../scheduler/completion-inbox.js'
import {
	GENAI,
	NAMZU,
	agentIterationSpanName,
	parentContext,
} from '../../../telemetry/attributes.js'
import { getTracer } from '../../../telemetry/runtime-accessors.js'
import { STRUCTURED_OUTPUT_TOOL_NAME } from '../../../tools/builtins/structuredOutput.js'
import type { CostInfo, TokenUsage } from '../../../types/common/index.js'
import { NamzuError } from '../../../types/errors/index.js'
import type { MessageId } from '../../../types/ids/index.js'
import {
	type Message,
	type UserMessage,
	createAssistantMessage,
	createRuntimeContextMessage,
	createSystemMessage,
} from '../../../types/message/index.js'
import { classifyProviderError } from '../../../types/provider/errors.js'
import type { ChatCompletionResponse } from '../../../types/provider/index.js'
import type { AnswerReview, AnswerReviewContext } from '../../../types/session/answer-review.js'
import type {
	SessionEvent,
	StepFailure,
	StepProvenance,
	StepResult,
	StopReason,
} from '../../../types/session/index.js'
import type { LLMToolSchema, ToolRegistryContract } from '../../../types/tool/index.js'
import { toErrorMessage } from '../../../utils/error.js'
import { stableDigest } from '../../../utils/hash.js'
import { generateMessageId } from '../../../utils/id.js'
import { createCallbackInference } from '../callback-inference.js'
import type { ToolCallOutcome } from '../executor.js'
import { projectObservationContext } from '../observation-context.js'
import { applyLifecycleHookResults } from '../plugin-hooks.js'
import {
	type RequestContextSnapshot,
	diffRequestContext,
	snapshotRequestContext,
} from '../request-context.js'
import {
	DEFAULT_MAX_REQUEST_RICH_CONTENT_BYTES,
	type RequestImageIdentity,
	markProviderRejectedImage,
	projectRequestRichContent,
} from '../request-rich-content.js'
import { formatSteeringNote, isOperatorUserMessage } from '../steering.js'
import { parseNativeCandidate } from './native-output.js'
import { holdForOutstandingWork, settleOutstandingWork } from './outstanding-work.js'
import { runAdvisoryPhase } from './phases/advisory.js'
import { runIterationCheckpoint } from './phases/checkpoint.js'
import {
	activeContextWindow,
	measureContext,
	relieveOverflow,
	runCompactionCheck,
} from './phases/compaction.js'
import type { IterationContext } from './phases/index.js'
import { runPlanGate } from './phases/plan.js'
import { runToolReview } from './phases/tool-review.js'
import { refreshWorkingMemory, splitWorkingMemoryForRequest } from './phases/working-memory.js'
import { streamWithProviderRejectedImageRecovery } from './provider-rejected-image.js'
import {
	type StepShaping,
	appendWorkContext,
	beforeStep,
	prepareStep,
	selectContextModel,
	stepContextMessage,
} from './step-shaping.js'
import { streamProviderTurn } from './stream-turn.js'

type ReviewRequest = Pick<AnswerReviewContext, 'requestMessages' | 'latestUserMessage'>

/** A host reviewer is not the model transport, even when its cause is an HTTP failure. */
class AnswerReviewFailure extends NamzuError {
	constructor(cause: unknown) {
		super({
			code: 'unknown',
			message: `Answer review failed: ${toErrorMessage(cause)}`,
			retryable: false,
			details: { phase: 'answer-review' },
			cause,
		})
	}
}
export type { IterationContext } from './phases/index.js'
export type { PhaseSignal } from './phases/index.js'
export type { ToolReviewOutcome } from './phases/index.js'

/**
 * How many times an answer may be handed back before the turn stops.
 *
 * Bounded for the same reason the structured-output re-prompt is: a judge
 * that never accepts would otherwise spend the whole token budget
 * rediscovering that, and the turn would end on a budget error rather than
 * on the thing that actually went wrong.
 */
const DEFAULT_ANSWER_REVIEW_LIMIT = 3

// Ending a turn changes the available actions, not the strength of its evidence.
// Use the same standard for warning closure and empty-completion recovery.
const CLOSING_RESPONSE_GUIDANCE =
	'Give a concise response using only what the available evidence supports. Attribute unverified statements to their source instead of presenting them as observed facts. If evidence is missing or conflicting, state what cannot be established. Do not claim unfinished work is complete. Do not request any more tool calls.'

export { awaitedJobGraceMs, settleGraceMs } from './outstanding-work.js'

export class IterationOrchestrator {
	private ctx: IterationContext
	private advisoryTurn:
		| {
				readonly iteration: number
				readonly requestMessages: readonly Message[]
				readonly response: Message
		  }
		| undefined

	/** Live only within its iteration; never joined by a guessed array offset. */
	getAdvisoryTurnContext():
		| import('../../../advisory/executor.js').AdvisoryTurnContext
		| undefined {
		const turn = this.advisoryTurn
		if (!turn || turn.iteration !== this.ctx.recorder.currentIteration) return undefined
		const start = this.ctx.recorder.messages.indexOf(turn.response)
		if (start < 0) return undefined
		return {
			iteration: turn.iteration,
			requestMessages: turn.requestMessages,
			subsequentMessages: this.ctx.recorder.messages.slice(start),
		}
	}
	/** Rejections so far. See {@link DEFAULT_ANSWER_REVIEW_LIMIT}. */
	private answerReviewAttempts = 0
	/**
	 * The last request envelope this turn recorded, so an unchanged one
	 * costs a hash and no event. Per RUNNER, not module-level: two turns in
	 * one process must not suppress each other's first envelope.
	 */
	private lastEnvelopeKey: string | undefined
	private previousRequestContext: RequestContextSnapshot | undefined

	private projectObservations(messages: Message[]): Message[] {
		const config = this.ctx.compactionConfig
		return config && config.strategy !== 'disabled' && config.deduplicateObservations !== false
			? projectObservationContext(messages, this.ctx.tools, config.preserveToolResultsFrom)
			: messages
	}
	/** Rich tool blocks already reported; durable history is scanned every turn. */
	private readonly warnedRichToolResults = new Set<string>()
	/**
	 * The previous iteration held a `stopWhen` decision open for a worker.
	 *
	 * Set when the stop predicate fired and the turn took one extra turn to
	 * read a delegated result, so the turn that then ends the turn can report
	 * WHY it is over. Without it the outcome was right and the record was
	 * wrong: the turn stopped because the host said so and reported `end_turn`,
	 * and this repo carries thirteen `StopReason` values precisely so that a
	 * run which ends for a nameable reason names it.
	 *
	 * Lives for exactly one iteration — see the read-and-clear at the top of
	 * the loop, which is the only site that touches it besides the one that
	 * sets it.
	 */
	private stopDeferredForOutstandingWork = false
	/** One current input, independent of the compactable history array. */
	private latestUserMessage: UserMessage | undefined

	constructor(ctx: IterationContext) {
		this.ctx = {
			...ctx,
			onSteeringDelivered: (text) => {
				this.rememberUserMessage(createRuntimeContextMessage(text, 'steering'))
			},
		}
		ctx.checkpointMgr.setLatestUserMessageSource(() => this.latestUserMessage)
		ctx.checkpointMgr.setAnswerReviewAttemptsSource?.(() => this.answerReviewAttempts)
		ctx.checkpointMgr.setStructuredReviewAttemptsSource?.(() => this.structuredReviewAttempts)
		ctx.checkpointMgr.setNativeStructuredAttemptsSource?.(() => this.nativeStructuredAttempts)
		if (ctx.structuredOutput?.mode === 'native') {
			const limit = ctx.structuredOutput.maxRetries
			if (limit !== undefined && (!Number.isSafeInteger(limit) || limit < 0))
				throw new RangeError(
					'Native structuredOutput.maxRetries must be a nonnegative safe integer',
				)
		}
		if (
			ctx.structuredOutput?.mode !== undefined &&
			!['tool', 'native'].includes(ctx.structuredOutput.mode)
		)
			throw new RangeError('Unknown structuredOutput.mode')
		const maxReviews = ctx.structuredOutput?.maxReviews
		if (maxReviews !== undefined && (!Number.isSafeInteger(maxReviews) || maxReviews < 0))
			throw new RangeError('structuredOutput.maxReviews must be a nonnegative safe integer')
		if (
			ctx.maxAnswerReviews !== undefined &&
			(!Number.isSafeInteger(ctx.maxAnswerReviews) || ctx.maxAnswerReviews < 0)
		)
			throw new RangeError('maxAnswerReviews must be a nonnegative safe integer')
	}

	/**
	 * Check the exact post-budget request for tool-result shapes the active driver
	 * cannot carry. Initial capability negotiation cannot see results produced by
	 * a later tool turn, so this boundary runs immediately before every provider
	 * call. Keys are durable call/block coordinates, which prevents old history
	 * from warning again on every subsequent iteration.
	 */
	private async reportUnsupportedToolResults(messages: readonly Message[]): Promise<void> {
		const capabilities =
			this.ctx.providerCapabilities ?? resolveProviderCapabilities(this.ctx.provider)
		const images: string[] = []
		const documents: string[] = []
		for (const message of messages) {
			if (message.role !== 'tool' || !Array.isArray(message.content)) continue
			for (const [index, block] of message.content.entries()) {
				const key = `${message.toolCallId}:${index}:${block.type}`
				if (this.warnedRichToolResults.has(key)) continue
				if (block.type === 'image' && !capabilities.supportsToolResultImages) {
					images.push(key)
				}
				if (block.type === 'document' && !capabilities.supportsToolResultDocuments) {
					documents.push(key)
				}
			}
		}

		const report = async (
			keys: readonly string[],
			capability: 'vision' | 'documents',
			label: 'image' | 'document',
		): Promise<void> => {
			if (keys.length === 0) return
			const message = `Provider '${this.ctx.provider.id}' declares it cannot map ${label} tool results, but this request carries ${keys.length} new ${label} block(s). The model will receive the driver's explicit text fallback instead of that content.`
			if (this.ctx.strictCapabilities) {
				throw new NamzuError({
					code: 'capability_unavailable',
					message,
					details: {
						providerId: this.ctx.provider.id,
						capability,
						blockCount: keys.length,
					},
				})
			}
			for (const key of keys) this.warnedRichToolResults.add(key)
			this.ctx.log.warn('Capability mismatch: the provider cannot map rich tool results', {
				'namzu.capability.detail': message,
				[GENAI.SYSTEM]: this.ctx.provider.id,
				'namzu.runtime.rich_tool_result_count': keys.length,
			})
			await this.ctx.emitEvent({
				type: 'capability_warning',
				turnId: this.ctx.recorder.turnId,
				capability,
				contentSource: 'tool-result',
				providerId: this.ctx.provider.id,
				message,
			})
		}

		await report(images, 'vision', 'image')
		await report(documents, 'documents', 'document')
	}

	/**
	 * Adopt the turn's span after construction.
	 *
	 * The orchestrator is built before `query()` enters its generator body,
	 * which is where the turn span is created — so the parent cannot be a
	 * constructor argument without reordering setup around one field.
	 */
	setRootSpan(span: Span): void {
		this.ctx = { ...this.ctx, rootSpan: span }
	}

	async *runLoop(): AsyncGenerator<SessionEvent> {
		const { turnConfig, recorder } = this.ctx
		const { model } = turnConfig
		const tracer = getTracer()
		// Resume hydration happens after construction, before the loop starts.
		this.latestUserMessage = this.ctx.checkpointMgr.restoredLatestUserMessage
		this.answerReviewAttempts = this.ctx.checkpointMgr.restoredAnswerReviewAttempts ?? 0
		this.structuredReviewAttempts = this.ctx.checkpointMgr.restoredStructuredReviewAttempts ?? 0
		this.nativeStructuredAttempts = this.ctx.checkpointMgr.restoredNativeStructuredAttempts ?? 0
		if (!this.latestUserMessage) {
			for (const message of recorder.messages) this.rememberUserMessage(message, false)
		}
		// The restored field can outlive the historical message it describes.
		// Only known post-checkpoint arrivals may supersede it, never an old
		// retained user turn encountered while scanning restored history.
		for (const message of this.ctx.resumedInput ?? []) this.rememberUserMessage(message)

		// What a recoverable failure pauses on when it lands before this
		// loop wrote a checkpoint of its own: without it a 429 on the first
		// request failed the turn, while the same 429 one request later
		// paused it. Taken after the intent above is known, so the
		// checkpoint names the message the turn is answering.
		await this.ctx.checkpointMgr.markLoopStart?.(recorder)

		// One context-overflow relief per *stuck point*, not per turn.
		//
		// The latch exists so that a second overflow immediately after a
		// successful compaction — meaning the prompt is irreducible — stops
		// instead of looping. It was never meant to disarm the mechanism for
		// the rest of the turn, which is what a turn-scoped flag did: one
		// relief at iteration 3 left iteration 40 to die on an overflow with
		// obvious moves left. It is cleared by a turn that actually
		// succeeded, which is the evidence that the turn is no longer stuck.
		let overflowRelieved = false

		const planSignal = yield* runPlanGate(this.ctx)
		if (planSignal === 'stop') return

		// A `finally` rather than a line at each exit, for the reason written
		// beside `iterSpan.end()` below: this loop leaves by eight `break`s,
		// two `return`s and a `throw`, and a rule every future edit has to
		// remember is a rule that gets forgotten — measured, it had been. Only
		// the ordinary final-answer exit consulted the inbox, so a turn that
		// ended on a terminal tool, a structured output or the host's
		// `stopWhen` settled over a finished worker's output and threw it away.
		// A `finally` also covers a generator abandoned by its consumer, which
		// no post-loop block reaches.
		try {
			while (true) {
				if (this.ctx.abortController.signal.aborted) {
					recorder.setStopReason('cancelled')
					recorder.markCancelled()
					break
				}
				if (
					this.ctx.structuredOutput?.mode === 'native' &&
					this.nativeStructuredAttempts > this.structuredOutputRetryLimit()
				) {
					recorder.setStopReason('structured_output_failed')
					break
				}
				if (
					this.ctx.reviewAnswer &&
					this.answerReviewAttempts > (this.ctx.maxAnswerReviews ?? DEFAULT_ANSWER_REVIEW_LIMIT)
				) {
					recorder.setStopReason('answer_rejected')
					break
				}
				if (
					this.ctx.structuredOutput?.review &&
					this.structuredReviewAttempts >
						(this.ctx.structuredOutput.maxReviews ?? DEFAULT_ANSWER_REVIEW_LIMIT)
				) {
					recorder.setStopReason('answer_rejected')
					break
				}
				// Read AND clear, in that order, in this one place.
				//
				// The flag is set by the previous iteration and read by this
				// one, so a clear that ran before the read would wipe it
				// before anything could use it — the obvious spelling of
				// "clear it at the top" is the broken one. Taking the value
				// into a local first gives the flag a lifetime of exactly one
				// iteration, which is the property that makes this cheap: no
				// path has to remember to clear it, because the next iteration
				// does so whether or not anything read it, and there is no
				// path by which a stale deferral can reach a later turn.
				const stopWasDeferredForOutstandingWork = this.stopDeferredForOutstandingWork
				this.stopDeferredForOutstandingWork = false

				const guardResult = this.ctx.guard.beforeIteration(
					recorder,
					this.ctx.abortController.signal,
				)

				if (guardResult.shouldStop) {
					if (guardResult.isCancelled) {
						this.ctx.log.info('Turn cancelled by signal', {
							[NAMZU.TURN_ID]: recorder.turnId,
						})
						recorder.setStopReason('cancelled')
						recorder.markCancelled()
						break
					}

					const stopReason = guardResult.stopReason ?? 'end_turn'
					this.ctx.log.info('Guard enforcing stop', {
						[NAMZU.TURN_ID]: recorder.turnId,
						'namzu.runtime.stop_reason': stopReason,
						[NAMZU.ITERATION]: recorder.currentIteration,
						'namzu.runtime.input_tokens': recorder.tokenUsage.promptTokens,
						'namzu.runtime.output_tokens': recorder.tokenUsage.completionTokens,
					})
					// A hard stop has no budget left for another model request.
					// Closing prose is requested at the warning threshold while
					// headroom remains; the completed work is already in history.
					recorder.setStopReason(stopReason)
					break
				}

				// Consulted here, after the guard and BEFORE the iteration is
				// counted or the provider is called. `stopWhen` reads `steps`
				// and so can only speak after the step it disliked has already
				// run and been paid for; this is the seam a host with a live
				// rate limit or a revoked tenant actually needs.
				const veto = await beforeStep(this.stepShaping(), recorder.currentIteration + 1)
				// The hook may settle because its turn signal was aborted. Stop
				// before interpreting that settlement as a policy refusal or
				// counting an iteration that will never reach the provider.
				if (this.ctx.abortController.signal.aborted) {
					recorder.setStopReason('cancelled')
					recorder.markCancelled()
					break
				}
				if (veto) {
					// Namespaced. The un-namespaced keys elsewhere in this file are
					// the frozen inventory LOG-22 exists to drain; a new call site
					// has no reason to join it.
					this.ctx.log.info('Step refused by beforeStep', {
						[NAMZU.TURN_ID]: recorder.turnId,
						[NAMZU.ITERATION]: recorder.currentIteration + 1,
						'namzu.step.veto_reason': veto.reason,
					})
					recorder.setLastError(`beforeStep refused the next step: ${veto.reason}`)
					recorder.setStopReason('step_refused')
					break
				}

				const forceFinalize = guardResult.forceFinalize
				const iterationNum = recorder.incrementIteration()
				this.ctx.log.debug('Iteration started', {
					[NAMZU.TURN_ID]: recorder.turnId,
					[NAMZU.ITERATION]: iterationNum,
					[GENAI.REQUEST_MODEL]: model,
					'namzu.runtime.force_finalize': forceFinalize,
					'namzu.runtime.message_count': recorder.messages.length,
				})

				const iterationActivity = this.ctx.activityStore.create({
					type: 'llm_turn',
					description: `LLM iteration ${iterationNum}`,
				})
				if (iterationActivity) {
					this.ctx.activityStore.start(iterationActivity.id)
				}

				// Parent explicitly: this body is an async generator, so the
				// ambient context at resume time belongs to the CONSUMER, not to
				// whoever created the turn span. Without this every iteration
				// emits as its own root and a 20-turn run shows up as 21
				// disconnected traces.
				const iterSpan = tracer.startSpan(
					agentIterationSpanName(iterationNum),
					{},
					parentContext(this.ctx.rootSpan),
				)

				// Everything the step record needs, hoisted so the `catch` can
				// read whatever the iteration got as far as computing.
				//
				// The failure path is the one the ledger's own argument was
				// written for and the one it never reached: an iteration that
				// threw recorded a span exception and re-threw, so the turn with
				// no record was exactly the turn that went wrong. A reader could
				// not tell that from a turn that never happened.
				//
				// Declared as `let` with real initial values rather than left
				// undefined, because a failure BEFORE the snapshot below is
				// taken has spent nothing, and these are then exact. The success
				// path is untouched: the assignments inside the try still happen
				// where they always did, so compaction and the working-memory
				// refresh stay outside a successful step's window.
				let stepStartedAt = Date.now()
				let usageBefore: TokenUsage = { ...recorder.tokenUsage }
				let costBefore: CostInfo = { ...recorder.costInfo }
				let stepModel = model
				let stepMessageId: MessageId | undefined
				let stepResponse: ChatCompletionResponse | undefined
				let stepServedBy: StepProvenance | undefined

				try {
					// Tool spans for this turn belong under this iteration. Inside
					// the try rather than before it: a throw from any of these left
					// the span open, and an iteration span that never ends is a
					// trace that never closes — the export is incomplete for exactly
					// the turn that failed.
					this.ctx.toolExecutor.setParentSpan(iterSpan)

					iterSpan.setAttributes({
						[NAMZU.ITERATION]: iterationNum,
						[GENAI.CONVERSATION_ID]: recorder.sessionId,
						[NAMZU.TURN_ID]: recorder.turnId,
						[GENAI.REQUEST_MODEL]: model,
					})

					await this.ctx.emitEvent({
						type: 'iteration_started',
						turnId: recorder.turnId,
						iteration: iterationNum,
					})
					yield* this.ctx.drainPending()

					if (this.ctx.pluginManager) {
						const hookResults = await this.ctx.pluginManager.executeHooks(
							'iteration_start',
							{
								sessionId: recorder.sessionId,
								turnId: recorder.turnId,
								iteration: iterationNum,
								signal: this.ctx.abortController.signal,
							},
							this.ctx.emitEvent,
						)
						applyLifecycleHookResults('iteration_start', hookResults)
						yield* this.ctx.drainPending()
					}

					// Re-pin the working-memory block from ground truth at the primacy
					// edge BEFORE compaction runs (so the refreshed slot is what
					// compaction preserves). No-op when no provider is configured.
					await refreshWorkingMemory(this.ctx)
					await runCompactionCheck(this.ctx)
					yield* this.ctx.drainPending()

					// Cache discipline: keep the tools param byte-stable even on the
					// forced-final iteration and forbid tool use via tool_choice
					// 'none' instead. Dropping the tools array would invalidate the
					// entire prompt-cache prefix (tools render at position 0) and
					// risks a 400 because the history still carries
					// tool_use/tool_result blocks.
					// Snapshot the cumulative counters so the step can report ITS
					// own usage rather than the turn total.
					stepStartedAt = Date.now()

					// Shape this step before calling the model. `stopWhen` decides
					// whether to keep going; this decides HOW. No-op when the host
					// supplied no hook.
					const contextModelBeforePreparation = this.ctx.contextModel ?? model
					const step = await prepareStep(this.stepShaping(), iterationNum)
					// Preparation inference belongs to the turn, not the main-model step.
					usageBefore = { ...recorder.tokenUsage }
					costBefore = { ...recorder.costInfo }
					stepModel = step.model ?? model
					await selectContextModel(this.stepShaping(), stepModel)
					// Preserve post-compaction preparation/recall semantics. A changed
					// model needs a second check against its own window; never replay
					// host preparation effects merely to rebuild its request guidance.
					if (stepModel !== contextModelBeforePreparation) await runCompactionCheck(this.ctx)
					// Publish context edits before entering a possibly slow provider.
					yield* this.ctx.drainPending()

					const stepAllowedTools = step.allowedTools ?? this.ctx.allowedTools
					const llmTools = this.ctx.tools
						.toLLMTools(stepAllowedTools)
						.filter(
							(tool) =>
								this.ctx.structuredOutput?.mode !== 'native' ||
								tool.function.name !== STRUCTURED_OUTPUT_TOOL_NAME,
						)
					// The same list the request was built from now also bounds what
					// may run. Narrowing only the request left the restriction
					// presentational — the model was shown fewer tools and could
					// still call any of them by name.
					this.ctx.toolExecutor.setStepAllowedTools(stepAllowedTools)
					const enforceToolInputSchema = enforcedModelInputToolNames(this.ctx.tools, llmTools)
					stepModel = step.model ?? model

					// The closing directive is appended LAST, after every piece of
					// request-only context below, so it is the final thing the model
					// reads on a forced-final step.
					const closingDirective = forceFinalize
						? createRuntimeContextMessage(
								`[SYSTEM] You are approaching your resource limits. ${CLOSING_RESPONSE_GUIDANCE}`,
								'limit-finalization',
							)
						: undefined
					const baseMessages = recorder.messages

					// Step guidance is appended to the REQUEST, never pushed onto
					// the turn's history: it applies to this step only, and pushing
					// it would accumulate one stale instruction per iteration.
					// Copy before it crosses the provider boundary. `recorder.messages`
					// is the LIVE run array, and the loop pushes onto it after the
					// call returns — so a driver that retains what it was handed
					// (to log it, cache it, or replay it on retry) watched its own
					// input grow new turns underneath it. A capture provider in the
					// estate recorded every turn as identical to the last for
					// exactly this reason. Shallow is enough: the defect is array
					// mutation, and per-iteration this is trivial next to the model
					// call it precedes.
					// A step's skills and its guidance ride the same ephemeral
					// system message. A driver may move it before history; changing
					// system guidance can therefore affect prefix caching. Observations
					// that need no system authority use step.context below.
					// `renderSkillsSection` already answers null for an empty list, so
					// there is no length check here — a second guard for the same
					// case is one more thing to keep in agreement with the first.
					const stepSkills = step.skills ? renderSkillsSection([...step.skills]) : null
					// A supervision change rides the same ephemeral slot, and for
					// the same reason: it applies to what happens next, not to the
					// run's history. The model plans around how closely it is being
					// watched — a turn that silently stops asking a human leaves it
					// batching destructive calls it expects to be reviewed, and one
					// that silently starts leaves it waiting on permission nobody
					// is left to give.
					//
					// Read-and-CLEAR, so it is said exactly once. Repeating it
					// every iteration would read as supervision moving again on
					// each turn.
					const policyChange = this.ctx.takeApprovalPolicyChange?.()
					const policyNotice = policyChange
						? `Approval policy changed from "${policyChange.from}" to "${policyChange.to}" (${policyChange.reason}). Tool calls from here on are reviewed under the new policy.`
						: null
					// State that changed during the turn, reported once per turn.
					// `turn` contributions are recomputed here, not fixed when the
					// run's prompt is assembled. They retain system authority and
					// may affect caching just like the other system contributions.
					const turnSections =
						this.ctx.promptContributions?.render('turn', {
							iteration: iterationNum,
						}) ?? []

					// Request-only context: observations that need no system
					// authority and change from request to request. They ride
					// runtime-context messages of kind `step-context` after the
					// history, which every driver keeps there and a caching driver
					// ends its breakpoint before — so a changed pin or a new turn
					// snapshot costs its own tokens, not a re-read of the history.
					// The working-memory slot keeps its place in the turn's history
					// (compaction preserves it there) and leaves the request's
					// system run here.
					const workingMemory = splitWorkingMemoryForRequest(baseMessages)
					const contextSections =
						this.ctx.promptContributions?.render('context', {
							iteration: iterationNum,
						}) ?? []

					const stepPreamble = [step.system, stepSkills, policyNotice, ...turnSections]
						.filter(Boolean)
						.join('\n\n')
					const requestHistory = stepPreamble
						? [...workingMemory.history, createSystemMessage(stepPreamble)]
						: [...workingMemory.history]
					if (workingMemory.context) requestHistory.push(workingMemory.context)
					if (contextSections.length > 0)
						requestHistory.push(stepContextMessage(contextSections.join('\n\n')))
					if (step.context) requestHistory.push(stepContextMessage(step.context))
					const messages = projectRequestRichContent(
						this.projectObservations(requestHistory),
						this.ctx.turnConfig.maxRequestRichContentBytes ??
							DEFAULT_MAX_REQUEST_RICH_CONTENT_BYTES,
					)
					appendWorkContext(this.stepShaping(), messages, iterationNum, step)
					if (closingDirective) messages.push(closingDirective)
					await this.reportUnsupportedToolResults(messages)
					yield* this.ctx.drainPending()

					// What the model is about to be ASKED, recorded when it
					// changed. `turn_started` carries one system prompt and tool
					// schemas never reached the transcript at all — while
					// `prepareStep` rewrites the system text, narrows the tool
					// list or swaps the model, and a step's skills ride the
					// ephemeral preamble above. So a transcript showed one
					// question for a turn that had asked several.
					//
					// Emitted only on a change: the digest is compared against
					// the last one this turn recorded, so the common case costs
					// one hash and nothing else. Copying an unchanged system
					// prompt every iteration is the fastest way to make a
					// durable log too large to read.
					const envelope = {
						model: stepModel,
						// Read off `messages`, which is what the request is actually
						// built from — including the ephemeral preamble. Recomputing
						// it from the turn's history would describe a request nobody
						// sent the moment the two diverge.
						systemPrompt: messages
							.filter((m) => m.role === 'system')
							.map((m) => String(m.content))
							.join('\n\n'),
						toolNames: llmTools.map((t) => t.function.name),
						// Over the SCHEMAS, sorted. A name list cannot see a tool
						// whose schema body changed while its name did not, which
						// is the change most likely to alter what the model does.
						toolSchemaDigest: stableDigest(
							[...llmTools].sort((a, b) => (a.function.name < b.function.name ? -1 : 1)),
						),
					}
					const envelopeKey = stableDigest(envelope)
					if (envelopeKey !== this.lastEnvelopeKey) {
						this.lastEnvelopeKey = envelopeKey
						await this.ctx.emitEvent?.({
							type: 'request_envelope',
							turnId: recorder.turnId,
							iteration: iterationNum,
							...envelope,
							toolNames: Object.freeze([...envelope.toolNames]),
						})
					}

					if (this.ctx.pluginManager) {
						const snapshot = snapshotRequestContext(messages)
						const context = Object.freeze({
							snapshot,
							...(this.previousRequestContext
								? {
										change: diffRequestContext(this.previousRequestContext, snapshot),
									}
								: {}),
						})
						this.previousRequestContext = snapshot
						const hookResults = await this.ctx.pluginManager.executeHooks(
							'pre_llm_call',
							{
								sessionId: recorder.sessionId,
								turnId: recorder.turnId,
								iteration: iterationNum,
								signal: this.ctx.abortController.signal,
								// Built inside the guard: a turn with no plugins installed
								// pays nothing for a projection nobody reads.
								request: Object.freeze({
									context,
									model: stepModel,
									// Copied per turn, not handed over live: these are the
									// run's own message objects, and a hook writing into
									// one would edit the history the turn is about to send.
									messages: Object.freeze(messages.map((m) => Object.freeze({ ...m }))),
									toolNames: Object.freeze(llmTools.map((t) => t.function.name)),
									temperature: step.temperature ?? turnConfig.temperature,
									maxTokens: step.maxResponseTokens ?? turnConfig.maxResponseTokens,
								}),
							},
							this.ctx.emitEvent,
						)
						applyLifecycleHookResults('pre_llm_call', hookResults)
						yield* this.ctx.drainPending()
					}

					// Phase 4 (ses_001-tool-stream-events): consume the
					// streaming response natively, emitting message and
					// tool-input lifecycle events as deltas arrive. The
					// helper yields SessionEvents through drainPending() so SSE
					// consumers see live progress; its return value is the
					// aggregated `ChatCompletionResponse` for the legacy
					// downstream paths (assistantMsg construction, working
					// state extraction, telemetry attribute stamping).
					//
					// The message id is minted HERE, immediately before the call
					// that announces it,
					// rather than inside that call. The return value never arrives
					// when the stream throws, so a step recorded from the catch
					// could otherwise never name the message — and a stream that
					// died part-way has already emitted both `message_started` and
					// `message_completed` under this id, which is the trail a
					// reader wants most on exactly that turn.
					stepMessageId = generateMessageId()
					const requestedMember = this.ctx.servingMember?.() ?? {
						index: 0,
						providerId: this.ctx.provider.id,
					}
					const requestedRoute: StepProvenance = {
						providerId: requestedMember.providerId,
						model: requestedMember.model ?? stepModel,
						chainIndex: requestedMember.index,
					}
					const operatorInputAtDispatch = this.latestUserMessage
					const latestReviewUserMessage =
						(this.ctx.reviewAnswer || this.ctx.structuredOutput?.review) && operatorInputAtDispatch
							? structuredClone(operatorInputAtDispatch)
							: undefined
					const { response, messageId, requestMessages } = yield* streamProviderTurn(
						this.ctx.provider,
						{
							model: stepModel,
							...(this.ctx.structuredOutput?.mode === 'native'
								? {
										responseFormat: {
											type: 'json_schema' as const,
											json_schema: {
												name: 'structured_output',
												schema: renderToolSchema(this.ctx.structuredOutput.schema),
												strict: true,
											},
										},
									}
								: {}),
							providerRoute: requestedRoute,
							messages,
							tools: llmTools.length > 0 ? llmTools : undefined,
							...(enforceToolInputSchema ? { enforceToolInputSchema } : {}),
							// The forced-final turn wins: a step that asked to force a
							// tool cannot override the loop's own decision to stop
							// asking for them. Otherwise the step's choice applies —
							// and only to this step, because the next one is prepared
							// from scratch.
							toolChoice:
								forceFinalize && llmTools.length > 0
									? 'none'
									: llmTools.length > 0
										? step.toolChoice
										: undefined,
							temperature: step.temperature ?? turnConfig.temperature,
							maxTokens: step.maxResponseTokens ?? turnConfig.maxResponseTokens,
							cacheControl: { type: 'auto' },
							...(turnConfig.thinking ? { thinking: turnConfig.thinking } : {}),
							...(turnConfig.effort ? { effort: turnConfig.effort } : {}),
							...(!forceFinalize && turnConfig.webSearch
								? { webSearch: turnConfig.webSearch }
								: {}),
							// Thread the turn abort into the model call so a Stop tears the
							// in-flight turn down (provider passes it to fetch; the consumer
							// also races it). Inert when never aborted.
							signal: this.ctx.abortController.signal,
						},
						this.ctx.emitEvent,
						this.ctx.drainPending,
						recorder.turnId,
						iterationNum,
						forceFinalize,
						this.ctx.log,
						iterSpan,
						stepMessageId,
						{
							onAccepted: (identity) => this.acceptProviderRejectedImage(identity),
						},
						Boolean(
							this.ctx.reviewAnswer || this.ctx.structuredOutput?.review || this.ctx.advisoryCtx,
						),
						recorder.sessionId,
					)
					stepResponse = response
					const reviewRequest: ReviewRequest = {
						...(requestMessages ? { requestMessages } : {}),
						...(latestReviewUserMessage ? { latestUserMessage: latestReviewUserMessage } : {}),
					}

					// Who answered THIS turn.
					//
					// The read is exact at this point and stays exact: a chain that
					// has produced output cannot fall over again inside the same
					// request, so the member at the cursor when the stream ends is
					// the one whose bytes are in `response`.
					//
					// Capture before host review: its auxiliary inference can move
					// the fallback cursor. Main-step usage and provenance must keep
					// naming the provider that produced this candidate.
					const servedBy: StepProvenance = ((): StepProvenance => {
						const member = this.ctx.servingMember?.() ?? {
							index: 0,
							providerId: this.ctx.provider.id,
						}
						return {
							providerId: member.providerId,
							// A member declared without a model asked for the model the
							// step named — which is what the decorator does with the
							// request, so this is a reading of it and not a guess.
							model: member.model ?? stepModel,
							chainIndex: member.index,
						}
					})()
					stepServedBy = servedBy

					// Main-loop turn: also records the prompt size compaction reads.
					//
					// `servedBy` is what prices it, and it is the exact pair —
					// the member at the cursor when the stream ended, and the
					// model it was asked for. This is the seam that ended the
					// always-zero cost: the rate lookup happens per turn,
					// against who actually answered, rather than against one
					// table the turn was constructed with.
					recorder.recordTurnUsage(response.usage, {
						providerId: servedBy.providerId,
						model: servedBy.model,
					})

					// The turn went through, so the turn is not sitting on an
					// irreducible prompt any more. Re-arm relief for the next one.
					overflowRelieved = false

					if (this.ctx.pluginManager) {
						const hookResults = await this.ctx.pluginManager.executeHooks(
							'post_llm_call',
							{
								sessionId: recorder.sessionId,
								turnId: recorder.turnId,
								iteration: iterationNum,
								signal: this.ctx.abortController.signal,
								response: Object.freeze({
									content: response.message.content,
									toolNames: Object.freeze(
										(response.message.toolCalls ?? []).map((c) => c.function.name),
									),
									finishReason: response.finishReason,
									usage: Object.freeze({ ...response.usage }),
								}),
							},
							this.ctx.emitEvent,
						)
						applyLifecycleHookResults('post_llm_call', hookResults)
						yield* this.ctx.drainPending()
					}

					this.ctx.log.debug('LLM response received', {
						[NAMZU.TURN_ID]: recorder.turnId,
						[NAMZU.ITERATION]: iterationNum,
						'namzu.runtime.finish_reason': response.finishReason,
						'namzu.runtime.has_content':
							response.message.content !== null && response.message.content.length > 0,
						'namzu.runtime.tool_call_count': response.message.toolCalls?.length ?? 0,
						[GENAI.USAGE_INPUT_TOKENS]: response.usage.promptTokens,
						[GENAI.USAGE_OUTPUT_TOKENS]: response.usage.completionTokens,
						'namzu.usage.total_tokens': recorder.tokenUsage.totalTokens,
						'namzu.runtime.total_cost': recorder.costInfo.totalCost,
					})

					// The context figures ride with the spend figures because a
					// surface showing one almost always wants the other — and
					// because the two were confusable enough that a host divided
					// cumulative spend by a context window and shipped it. They
					// are measured here rather than left to be derived, since the
					// only correct derivation needs internals a host cannot see.
					//
					// Absent when the turn has no compaction config: nothing then
					// resolves a window, and inventing one would be the guess this
					// replaces.
					const contextFigures = this.ctx.compactionConfig
						? (() => {
								const measured = measureContext(this.ctx)
								const window = activeContextWindow(this.ctx)
								return {
									contextTokens: measured.tokens,
									contextMeasuredBy: measured.source,
									contextWindowTokens: window.tokens,
									windowSource: window.source,
								}
							})()
						: {}

					await this.ctx.emitEvent({
						type: 'token_usage_updated',
						turnId: recorder.turnId,
						usage: recorder.tokenUsage,
						budget: recorder.budget?.summary(),
						cost: recorder.costInfo,
						...contextFigures,
					})

					// Durable reasoning and its adapter-private replay envelope ride
					// with the turn they belong to. Trimming therefore removes both;
					// retaining them gives the target adapter enough evidence to
					// validate native replay against the exact serving route.
					const assistantMsg = createAssistantMessage(
						response.message.content,
						forceFinalize ? undefined : response.message.toolCalls,
						response.message.reasoning,
						// Rides with the turn it belongs to, like reasoning does, so
						// trimming or compacting the turn takes its evidence with it
						// rather than leaving citations pointing at prose that is gone.
						response.message.citations,
						{
							type: 'model',
							...servedBy,
							...(response.message.replayState !== undefined
								? { replayState: response.message.replayState }
								: {}),
						},
						response.message.textParts,
					)
					recorder.pushMessage(assistantMsg)
					if (this.ctx.advisoryCtx && requestMessages) {
						this.advisoryTurn = { iteration: iterationNum, requestMessages, response: assistantMsg }
					}

					if (this.ctx.workingStateManager && this.ctx.compactionConfig && assistantMsg.content) {
						extractFromAssistantMessage(
							this.ctx.workingStateManager,
							assistantMsg.content,
							this.ctx.compactionConfig,
						)
					}

					yield* this.ctx.drainPending()

					iterSpan.setAttributes({
						[GENAI.USAGE_INPUT_TOKENS]: response.usage.promptTokens,
						[GENAI.USAGE_OUTPUT_TOKENS]: response.usage.completionTokens,
					})
					iterSpan.setStatus({ code: SpanStatusCode.OK })

					if (iterationActivity) {
						this.ctx.activityStore.complete(iterationActivity.id, {
							content: response.message.content,
							hasToolCalls: forceFinalize ? false : !!response.message.toolCalls?.length,
						})
					}

					// Tool calls beat the finish reason. The reason is the
					// provider's SUMMARY of the turn and the tool calls are the
					// turn itself, so when they disagree the calls are the fact.
					// Several function-calling endpoints — gateways and local servers
					// especially — report `stop` alongside a populated
					// `tool_calls`, and three of this repo's drivers pass that
					// value through untouched.
					//
					// Reading `stop` first meant the turn ended with every
					// requested call silently skipped, an assistant message
					// carrying tool_use blocks that were never answered, and a
					// run that settled `end_turn` having done nothing it was
					// asked to do. Checking the calls first costs nothing when
					// the provider is honest and is the only thing that saves the
					// run when it is not.
					const hasToolCalls = (response.message.toolCalls?.length ?? 0) > 0

					if (forceFinalize || !hasToolCalls) {
						// Every task-dispatch tool (create_task, continue_task, Agent)
						// is BLOCKING: the worker's output returns as the dispatching
						// tool_use's canonical tool_result, so by the time the model
						// ends its turn nothing launched by this turn should still be
						// in flight. A running task here is an orphan (interrupted
						// tool execution, cancel race) with no delivery path back to
						// the parent — the <task-notification> producer was removed
						// in dc16d58, so waiting on the queue could only ever time
						// out. Log the orphans honestly and end the turn normally.
						if (!forceFinalize && this.hasRunningAgentTasks()) {
							this.ctx.log.warn(
								'LLM ended turn with agent tasks still running — ending turn without waiting (orphan tasks have no delivery path)',
								{
									[NAMZU.TURN_ID]: recorder.turnId,
									[NAMZU.ITERATION]: iterationNum,
								},
							)
						}

						// This iteration gets a step too.
						//
						// It did not, and the ledger's own contract said it should:
						// `StepResult` is documented as "what one iteration of the
						// agent loop did" and `stepNumber` as "1-based, matching
						// `iteration` on the turn events". Every path below emits
						// `iteration_completed` with this iteration's number, and
						// none of them recorded a step — so the events said
						// iteration N happened and `steps` had no entry N. The
						// invariant was not a definition anyone chose; it was
						// already false.
						//
						// What it cost: measured on a two-iteration run, one tool
						// call then an answer, 220 of 330 tokens belonged to no
						// step. That is not a rounding error and it is structurally
						// the worst turn to lose — the answering turn carries the
						// largest prompt, so the unattributed share GROWS with
						// context length.
						//
						// Recorded HERE, at the top of the branch, rather than at
						// each of its exits. Every path out of this block is a
						// `continue`, a `break` or a `return`, so one call covers
						// the terminal answer, the forced-final summary, the
						// auto-continuation, the structured-output re-prompt and the
						// answer-review rejection — all of which spend a turn's
						// tokens. Placing it at the exits instead would be five call
						// sites to keep in agreement, and the one added later would
						// be the one that got missed.
						//
						// No tool results, because this branch is defined by their
						// absence. `toolExecutionMs` is 0 for the same reason.
						//
						// A step is an ITERATION'S MAIN TURN, and side calls are
						// still not steps — the compaction verifier, the advisory
						// executor, and the empty-completion retry a few lines below
						// all spend tokens inside an iteration without being one.
						// Their usage reaches `run.tokenUsage` and no step, so the
						// ledger reconciles with the turn total for a turn that makes
						// no side calls and undercounts by exactly those calls for a
						// run that does. That residual is named rather than fixed
						// here: attributing a side call needs a record that is not a
						// step, which is a different claim.

						this.recordStep({
							stepNumber: iterationNum,
							model: stepModel,
							servedBy,
							messageId,
							response,
							toolResults: [],
							toolExecutionMs: 0,
							startedAt: stepStartedAt,
							usageBefore,
							costBefore,
						})

						if (this.ctx.structuredOutput?.mode === 'native') {
							const candidate = await parseNativeCandidate(
								this.ctx.structuredOutput.schema,
								response,
								this.ctx.abortController.signal,
							)
							let outcome: 'accepted' | 'retry' | 'exhausted' | 'cancelled'
							if (candidate.success)
								outcome = await this.reviewStructuredOutput(
									candidate.value,
									reviewRequest,
									stepModel,
								)
							else {
								this.nativeStructuredAttempts++
								recorder.pushMessage(
									createRuntimeContextMessage(
										'Return a complete JSON value matching the supplied response schema. Do not continue a partial JSON fragment.',
										'structured-output',
									),
								)
								const checkpoint = await this.ctx.checkpointMgr.create(recorder, iterationNum)
								await this.ctx.emitEvent({
									type: 'checkpoint_created',
									turnId: recorder.turnId,
									checkpointId: checkpoint.id,
									iteration: iterationNum,
								})
								outcome =
									this.nativeStructuredAttempts > this.structuredOutputRetryLimit()
										? 'exhausted'
										: 'retry'
							}
							await this.ctx.emitEvent({
								type: 'iteration_completed',
								turnId: recorder.turnId,
								iteration: iterationNum,
								hasToolCalls: false,
							})
							yield* this.ctx.drainPending()
							if (this.ctx.abortController.signal.aborted || outcome === 'cancelled') {
								recorder.setStopReason('cancelled')
								recorder.markCancelled()
								break
							}
							if (outcome === 'accepted') {
								if (!forceFinalize) {
									const changed = yield* holdForOutstandingWork(this.ctx, iterationNum, false, () =>
										this.deliverInbound(),
									)
									const inbound = this.deliverInbound()
									if (changed || inbound > 0) continue
								}
								if (this.ctx.abortController.signal.aborted) {
									recorder.setStopReason('cancelled')
									recorder.markCancelled()
									break
								}
								this.publishStructuredOutput()
								recorder.setStopReason('end_turn')
								break
							}
							if (outcome === 'exhausted') {
								recorder.setStopReason(
									candidate.success ? 'answer_rejected' : 'structured_output_failed',
								)
								break
							}
							continue
						}

						const hasContent =
							response.message.content !== null && response.message.content.length > 0

						// Auto-continuation on `stop_reason: max_tokens`. The
						// model hit its per-call output cap mid-text (NOT
						// mid-tool-use — that path is handled separately
						// below via `inputTruncated`). Push a synthetic
						// "continue" user message and let the loop fire
						// another turn. The provider receives the partial
						// assistant content + the continue prompt and
						// resumes from where it left off, mirroring the
						// Auto-continuation after an output-ceiling cutoff.
						//
						// Guards:
						//   - `hasContent` so we don't loop forever on an
						//     empty cutoff (a provider occasionally emits
						//     `stop_reason: max_tokens` with no content
						//     when an injected pre-fill blocks the model).
						//   - `!forceFinalize` so the forced-finalize path
						//     never auto-continues — that path is invoked
						//     specifically to extract a closing summary.
						//   - max_iterations bounds the loop in any case.
						if (!forceFinalize && response.finishReason === 'length' && hasContent) {
							this.ctx.log.info('LLM hit max_tokens mid-text — auto-continuing', {
								[NAMZU.TURN_ID]: recorder.turnId,
								[NAMZU.ITERATION]: iterationNum,
								[GENAI.USAGE_OUTPUT_TOKENS]: response.usage.completionTokens,
							})
							recorder.pushMessage(
								createRuntimeContextMessage(AUTO_CONTINUATION_USER_MESSAGE, 'auto-continuation'),
							)
							await this.ctx.emitEvent({
								type: 'iteration_completed',
								turnId: recorder.turnId,
								iteration: iterationNum,
								hasToolCalls: false,
							})
							yield* this.ctx.drainPending()
							continue
						}

						// The model tried to finish in prose while a structured
						// output was demanded. Send it back with the schema error
						// rather than returning an unusable result — this is the
						// re-prompt half, and it is bounded so a model that cannot
						// satisfy the schema fails loudly instead of looping.
						if (!forceFinalize && this.needsStructuredOutput()) {
							const attempt = ++this.structuredOutputAttempts
							const limit = this.structuredOutputRetryLimit()
							if (attempt > limit) {
								this.ctx.log.warn('Structured output not produced within its retries', {
									[NAMZU.TURN_ID]: recorder.turnId,
									'namzu.runtime.attempts': attempt - 1,
								})
								recorder.setStopReason('structured_output_failed')
								break
							}
							this.ctx.log.info('Re-prompting for structured output', {
								[NAMZU.TURN_ID]: recorder.turnId,
								'namzu.retry.attempt': attempt,
								'namzu.runtime.limit': limit,
							})
							recorder.pushMessage(
								createRuntimeContextMessage(STRUCTURED_OUTPUT_REPROMPT, 'structured-output'),
							)
							await this.ctx.emitEvent({
								type: 'iteration_completed',
								turnId: recorder.turnId,
								iteration: iterationNum,
								hasToolCalls: false,
							})
							yield* this.ctx.drainPending()
							continue
						}

						// Let the host judge the ANSWER and hand back work.
						//
						// The stop predicate is only consulted after tools ran, so
						// there was no seam here at all: the moment the model
						// stopped calling tools the turn finalized, whatever it had
						// produced. Verify-then-fix — run the build, feed the
						// failure back, let it try again — meant starting a whole
						// new turn and re-supplying the context the first one had.
						//
						// Shaped after the structured-output re-prompt directly
						// above, which solves the same problem for one specific
						// judge: bounded attempts, feedback as a user message, and
						// a loud stop rather than a loop.
						if (!forceFinalize && this.ctx.reviewAnswer) {
							const review = await this.reviewAnswer(
								response.message.content ?? '',
								reviewRequest,
								stepModel,
							)
							if (this.ctx.abortController.signal.aborted) {
								recorder.setStopReason('cancelled')
								recorder.markCancelled()
								break
							}
							if (review && !review.accept) {
								const attempt = ++this.answerReviewAttempts
								recorder.pushMessage(createRuntimeContextMessage(review.feedback, 'answer-review'))
								// Commit the consumed allowance with its feedback before another
								// request, including exhaustion. Compaction cannot reset this quota.
								const checkpoint = await this.ctx.checkpointMgr.create(recorder, iterationNum)
								await this.ctx.emitEvent({
									type: 'checkpoint_created',
									turnId: recorder.turnId,
									checkpointId: checkpoint.id,
									iteration: iterationNum,
								})
								if (this.ctx.abortController.signal.aborted) {
									recorder.setStopReason('cancelled')
									recorder.markCancelled()
									break
								}
								const limit = this.ctx.maxAnswerReviews ?? DEFAULT_ANSWER_REVIEW_LIMIT
								if (attempt > limit) {
									this.ctx.log.warn('Answer rejected more times than the turn allows', {
										[NAMZU.TURN_ID]: recorder.turnId,
										'namzu.runtime.attempts': attempt - 1,
										'namzu.runtime.limit': limit,
									})
									recorder.setStopReason('answer_rejected')
									break
								}
								this.ctx.log.info('Answer rejected — returning it to the model', {
									[NAMZU.TURN_ID]: recorder.turnId,
									'namzu.retry.attempt': attempt,
									'namzu.runtime.limit': limit,
								})
								await this.ctx.emitEvent({
									type: 'iteration_completed',
									turnId: recorder.turnId,
									iteration: iterationNum,
									hasToolCalls: false,
								})
								yield* this.ctx.drainPending()
								continue
							}
						}

						// A background worker is still out there, and this turn was
						// about to end the turn.
						//
						// Settling here would throw away the very thing the launch
						// existed to produce: the supervisor said "launched", the
						// worker had not finished, and the turn closed over it.
						if (
							!forceFinalize &&
							(yield* holdForOutstandingWork(this.ctx, iterationNum, false, () =>
								this.deliverInbound(),
							))
						) {
							continue
						}

						// Anything queued while this turn ran, on the path where
						// there is no tool result to carry it. Without this the
						// run settles with the channel still pending — which is
						// the failure the steering channel's own test used to
						// PIN as correct behaviour.
						//
						// After the outstanding-work hold above, so a delivery
						// does not race a worker still finishing, and before the
						// settle below, which is the last moment it can matter.
						if (!forceFinalize && this.deliverInbound() > 0) {
							await this.ctx.emitEvent({
								type: 'iteration_completed',
								turnId: recorder.turnId,
								iteration: iterationNum,
								hasToolCalls: false,
							})
							yield* this.ctx.drainPending()
							continue
						}

						// A limit-requested summary bypasses prose review and further
						// work. Preserve that limit on settlement, even if the provider
						// reports a normal text completion and headroom still remains.
						let closingStopReason: StopReason | undefined = forceFinalize
							? guardResult.stopReason
							: undefined
						if (!hasContent && !forceFinalize) {
							this.ctx.log.warn('Empty completion detected — requesting final summary', {
								[NAMZU.ITERATION]: iterationNum,
								'namzu.runtime.finish_reason': response.finishReason,
							})
							closingStopReason = await this.requestFinalResponse(model, 'end_turn')
							yield* this.ctx.drainPending()
						}

						await this.ctx.emitEvent({
							type: 'iteration_completed',
							turnId: recorder.turnId,
							iteration: iterationNum,
							hasToolCalls: false,
						})
						yield* this.ctx.drainPending()
						// A Stop that lands AFTER the final turn streamed but before
						// this break must settle the turn as cancelled, not end_turn —
						// otherwise the just-produced answer is recorded as a clean
						// completion. Mirrors the between-iteration cancel at :511.
						if (this.ctx.abortController.signal.aborted) {
							recorder.setStopReason('cancelled')
							recorder.markCancelled()
							break
						}
						// The host's stop predicate, if the previous turn deferred it
						// to let the model read a delegated result. That extra turn
						// is prose, and `stopWhen` is consulted only after a tool
						// batch, so the predicate is never asked again — reporting
						// `end_turn` would name the shape of the last message rather
						// than the reason the turn is over.
						//
						// Only here. A terminal tool and a captured structured output
						// also settle as `end_turn`, and there the deferred predicate
						// is not why the turn ended: those decided the answer
						// themselves.
						recorder.setStopReason(
							closingStopReason ??
								(stopWasDeferredForOutstandingWork ? 'stop_condition' : 'end_turn'),
						)
						break
					}

					const reviewOutcome = yield* runToolReview(this.ctx, response, iterationNum)

					// The step record is built even for a rejected batch: a turn that
					// spent a turn getting its tools refused still spent the tokens,
					// and a caller reconstructing cost per step must see it.
					this.recordStep({
						stepNumber: iterationNum,
						// The model this step ASKED for. It used to be `model`, the
						// run's own — so a `prepareStep` that routed one step to a
						// cheaper model was recorded as the expensive one, with no
						// provider chain involved.
						model: stepModel,
						servedBy,
						messageId,
						response,
						toolResults: reviewOutcome.results,
						toolExecutionMs: reviewOutcome.durationMs,
						startedAt: stepStartedAt,
						usageBefore,
						costBefore,
					})

					if (reviewOutcome.decision === 'stop') {
						return
					}

					if (reviewOutcome.decision === 'rejected') {
						continue
					}

					// A successful `structured_output` call IS the answer, so the
					// run ends here rather than paying for another turn whose only
					// job would be to restate it — unless it shared its turn with
					// other calls, which relays instead. See the method.
					const structuredOutcome = await this.captureStructuredOutput(
						reviewOutcome.results,
						response,
						reviewRequest,
						stepModel,
					)
					if (
						structuredOutcome === 'retry' ||
						structuredOutcome === 'exhausted' ||
						structuredOutcome === 'cancelled'
					) {
						await this.ctx.emitEvent({
							type: 'iteration_completed',
							turnId: recorder.turnId,
							iteration: iterationNum,
							hasToolCalls: true,
						})
						yield* this.ctx.drainPending()
						if (this.ctx.abortController.signal.aborted) {
							recorder.setStopReason('cancelled')
							recorder.markCancelled()
							break
						}
						if (structuredOutcome === 'retry') continue
						recorder.setStopReason(
							structuredOutcome === 'cancelled' ? 'cancelled' : 'answer_rejected',
						)
						if (structuredOutcome === 'cancelled') recorder.markCancelled()
						break
					}
					if (structuredOutcome === 'accepted') {
						await this.ctx.emitEvent({
							type: 'iteration_completed',
							turnId: recorder.turnId,
							iteration: iterationNum,
							hasToolCalls: true,
						})
						yield* this.ctx.drainPending()
						if (this.ctx.abortController.signal.aborted) {
							recorder.setStopReason('cancelled')
							recorder.markCancelled()
							break
						}
						if (!forceFinalize) {
							const inbound = this.deliverInbound()
							// Tool-result steering may already have been delivered by
							// runToolReview. Its candidate still answers the older input.
							if (inbound > 0 || this.latestUserMessage !== operatorInputAtDispatch) continue
						}
						if (this.ctx.abortController.signal.aborted) {
							recorder.setStopReason('cancelled')
							recorder.markCancelled()
							break
						}
						this.ctx.log.info('Structured output produced — ending turn', {
							[NAMZU.TURN_ID]: recorder.turnId,
							[NAMZU.ITERATION]: iterationNum,
						})
						this.publishStructuredOutput()
						recorder.setStopReason('end_turn')
						break
					}

					// A tool the author declared terminal settles the turn with its
					// own output, the same rule `structured_output` has always
					// had. Without it a delegation cost the parent one more model
					// call at full context whose only job was to restate what the
					// worker already said — and to restate it through the parent's
					// compacted view, so the caller did not even receive the
					// worker's words.
					const settled = this.terminalToolOutput(reviewOutcome.results, response)
					if (settled !== undefined) {
						this.ctx.log.info('Terminal tool produced the answer — ending turn', {
							[NAMZU.TURN_ID]: recorder.turnId,
							[NAMZU.ITERATION]: iterationNum,
							[GENAI.TOOL_NAME]: settled.toolName,
						})
						recorder.setResult(settled.output, 'review')
						recorder.setStopReason('end_turn')
						await this.ctx.emitEvent({
							type: 'iteration_completed',
							turnId: recorder.turnId,
							iteration: iterationNum,
							hasToolCalls: true,
						})
						yield* this.ctx.drainPending()
						break
					}

					// Evaluated AFTER the tools ran, so a predicate can see what they
					// returned — which is what makes a terminal submit_answer tool
					// usable without discarding its output.
					if (await this.shouldStop()) {
						// Outstanding work outranks the host's stop predicate —
						// a delegated task the completion inbox is expecting, or
						// a background job the model told `wait_for_job` it is
						// waiting on.
						//
						// This is a precedence rule chosen here, not something
						// `stopWhen` implies — a stop predicate is a programmable
						// halt and says nothing about whether the answer is
						// complete, which is what separates it from a terminal
						// tool or a captured structured output. Those decide the
						// result, so no turn follows and a hold would buy nothing.
						// This one only says "stop", and stopping one turn later
						// with the result in hand is a better reading of the
						// host's intent than stopping now and discarding it.
						//
						// Bounded by what is left to deliver, not by a count.
						// Each delivery consumes what it delivered — the inbox is
						// drained, and a job exit's notice is taken with the
						// record of the exits it accounts for — so the predicate
						// is asked again next turn against whatever is still
						// outstanding. One task deferred it once; two awaited
						// jobs exiting a minute apart defer it twice, each time
						// for a turn the model spends on news it has not read.
						// `maxIterations` and the turn's own deadline bound all of
						// it regardless, and a leg with nothing pending never
						// opens a hold at all.
						if (
							yield* holdForOutstandingWork(this.ctx, iterationNum, true, () =>
								this.deliverInbound(),
							)
						) {
							// Remember WHY the next turn exists, so the turn that
							// ends the turn can name the host's decision instead of
							// reporting the shape of the last message.
							this.stopDeferredForOutstandingWork = true
							continue
						}

						this.ctx.log.info('Stop condition met', {
							[NAMZU.TURN_ID]: recorder.turnId,
							[NAMZU.ITERATION]: iterationNum,
						})
						recorder.setStopReason('stop_condition')
						await this.ctx.emitEvent({
							type: 'iteration_completed',
							turnId: recorder.turnId,
							iteration: iterationNum,
							hasToolCalls: true,
						})
						yield* this.ctx.drainPending()
						break
					}

					const checkpointSignal = yield* runIterationCheckpoint(this.ctx, iterationNum)
					if (checkpointSignal === 'stop') {
						return
					}

					// Workers that finished with nobody listening.
					//
					// A completion normally reaches the supervisor as the
					// `tool_result` of the `create_task` that launched it. Two
					// cases have no such call: a launch made in the background on
					// purpose, and a blocking launch whose deadline passed — the
					// model was told "timed out, it may still be running" and the
					// worker then finished, holding a result nothing would read.
					//
					// This is the channel that was removed in `dc16d58` because it
					// double-delivered: it fired for completions the blocking tool
					// had already handed over, so the supervisor saw each result
					// twice. The inbox restores it with the distinction that was
					// missing — a tool that delivers a completion claims it, and
					// only unclaimed ones arrive here.
					//
					// Placed beside the advisory phase deliberately: that is the
					// established seam for putting a user message in after tool
					// results and before the next turn.
					const unheard = this.ctx.completionInbox?.drain() ?? []
					if (unheard.length > 0) {
						this.ctx.log.info('Delivering unawaited task completions', {
							[NAMZU.TURN_ID]: recorder.turnId,
							[NAMZU.ITERATION]: iterationNum,
							'namzu.runtime.tasks': unheard.map((h) => h.taskId),
						})
						recorder.pushMessage(
							createRuntimeContextMessage(formatCompletionNotification(unheard), 'task-completion'),
						)
					}

					// The same seam, for the two channels that could accept text
					// and never deliver it. Placed here rather than at the top of
					// the next iteration so a message queued during THIS turn is
					// in the history the next request is built from.
					this.deliverInbound()

					await runAdvisoryPhase(this.ctx, iterationNum, response, this.getAdvisoryTurnContext())

					if (this.ctx.pluginManager) {
						const hookResults = await this.ctx.pluginManager.executeHooks(
							'iteration_end',
							{
								sessionId: recorder.sessionId,
								turnId: recorder.turnId,
								iteration: iterationNum,
								signal: this.ctx.abortController.signal,
							},
							this.ctx.emitEvent,
						)
						applyLifecycleHookResults('iteration_end', hookResults)
						yield* this.ctx.drainPending()
					}

					await this.ctx.emitEvent({
						type: 'iteration_completed',
						turnId: recorder.turnId,
						iteration: iterationNum,
						hasToolCalls: true,
					})
					yield* this.ctx.drainPending()
				} catch (err) {
					const cancelled = this.ctx.abortController.signal.aborted

					// This iteration gets a step too, and it is the one the
					// argument three hundred lines above was actually about.
					//
					// That docblock makes the case for a rejected tool batch — "a
					// run that spent a turn getting its tools refused still spent
					// the tokens" — and every call site it produced sat on a
					// success path. So the ledger was complete except on the turns
					// that failed, which is the worst shape it could have: an
					// evidence record that goes quiet exactly where something went
					// wrong reads as "nothing went wrong". A reader could not
					// distinguish iteration N failing from iteration N never
					// happening, while the events said plainly that it started.
					//
					// Recorded HERE, at the top of the catch, rather than at each
					// of its exits — the same reasoning the success path already
					// wrote down for itself. All three exits spend a turn: the
					// cancellation breaks, the overflow-relief retry continues
					// under a NEW iteration number (so its tokens belong to no
					// later step), and the re-throw ends the turn.
					//
					// What it carries is what the iteration got as far as knowing.
					// `usage` is the same subtraction a successful step makes, so
					// a turn that failed after the provider answered carries that
					// answer's tokens, and one that failed before it carries the
					// zero it actually spent. Nothing is estimated to fill a gap.
					//
					// At most ONE step per iteration. Both success paths record
					// before the work that follows them — the advisory phase, the
					// structured-output capture, the `iteration_end` hooks, the
					// terminal `iteration_completed` — and any of those can throw
					// into here. A second entry numbered N would double-count that
					// turn's tokens against `run.tokenUsage`, which is the same
					// class of wrong as dropping them and harder to notice, since
					// the ledger would look fuller rather than emptier. That turn's
					// own verdict is already written down; the failure that
					// followed it reaches the caller as the turn's error.
					if (this.steps.at(-1)?.stepNumber === iterationNum) {
						this.ctx.log.warn('Iteration failed after its step was already recorded', {
							[NAMZU.TURN_ID]: recorder.turnId,
							[NAMZU.ITERATION]: iterationNum,
							'exception.message': toErrorMessage(err),
						})
					} else {
						this.recordStep({
							stepNumber: iterationNum,
							model: stepModel,
							...(stepServedBy ? { servedBy: stepServedBy } : {}),
							...(stepMessageId ? { messageId: stepMessageId } : {}),
							...(stepResponse ? { response: stepResponse } : {}),
							// Tool outcomes are produced and returned together by
							// `runToolReview`, so a throw from inside it leaves none
							// to salvage: an empty list here means "none came back",
							// which is what the shorter-than-`toolCalls` contract
							// says.
							toolResults: [],
							toolExecutionMs: 0,
							startedAt: stepStartedAt,
							usageBefore,
							costBefore,
							unfinished: cancelled
								? { finishReason: 'cancelled' }
								: {
										finishReason: 'error',
										failure: describeStepFailure(err, this.ctx.provider.id),
									},
						})
					}

					// A Stop that aborted the in-flight turn surfaces here as a
					// thrown abort (the provider stream was raced against the turn
					// signal). Settle it as a CANCELLATION — mirroring the
					// between-iteration cancel at the top of the loop — rather than
					// recording it as an SDK failure (error span + failed activity)
					// and re-throwing. The turn then returns cleanly with a
					// 'cancelled' stop reason instead of propagating an error.
					if (cancelled) {
						recorder.setStopReason('cancelled')
						recorder.markCancelled()
						break
					}

					// The one provider failure the kernel can actually do something
					// about. `context_length_exceeded` is correctly non-retryable —
					// resending the identical prompt cannot help — but the kernel
					// owns a compaction subsystem that can make the prompt smaller.
					// Without this the turn died holding the remedy: the threshold
					// path had simply guessed low, which a turn carrying images or a
					// language the chars-per-token ratio does not fit will do.
					//
					// Relief is attempted ONCE per iteration and only when it
					// actually shed something. A second overflow after a successful
					// compaction means the prompt is irreducible, and looping on it
					// would burn the budget to arrive at the same error.
					if (
						!overflowRelieved &&
						!(err instanceof AnswerReviewFailure) &&
						classifyProviderError(err, this.ctx.provider.id).code === 'context_length_exceeded'
					) {
						overflowRelieved = true
						const shed = await relieveOverflow(this.ctx)
						if (shed) {
							this.ctx.log.info('Retrying the turn after relieving a context overflow', {
								[NAMZU.TURN_ID]: recorder.turnId,
								[NAMZU.ITERATION]: iterationNum,
							})
							if (iterationActivity) {
								this.ctx.activityStore.complete(iterationActivity.id)
							}
							continue
						}
					}

					if (iterationActivity) {
						this.ctx.activityStore.fail(iterationActivity.id, toErrorMessage(err))
					}

					iterSpan.setStatus({
						code: SpanStatusCode.ERROR,
						message: toErrorMessage(err),
					})
					iterSpan.recordException(err instanceof Error ? err : new Error(String(err)))
					throw err
				} finally {
					this.advisoryTurn = undefined
					// The only place the iteration span ends. It used to be ended at each of
					// seventeen exits, which is a rule every future edit has to
					// remember; a generator abandoned by its consumer never reached
					// any of them.
					iterSpan.end()
				}
			}
		} finally {
			settleOutstandingWork(this.ctx)
		}
	}

	/**
	 * The context and the two live reads the step-shaping helpers share.
	 *
	 * Built per call rather than held: `latestUserMessage` is replaced on
	 * every operator turn and `steps` grows by one per step, so a captured
	 * value would describe an earlier return.
	 */
	private stepShaping(): StepShaping {
		return {
			ctx: this.ctx,
			latestUserMessage: () => this.latestUserMessage,
			steps: () => this.steps,
		}
	}

	/** Steps completed so far, exposed on the returned `Turn`. */
	private readonly steps: StepResult[] = []

	getSteps(): readonly StepResult[] {
		return this.steps
	}

	/**
	 * Fold one iteration into a `StepResult`.
	 *
	 * Every field here was already computed somewhere in the loop; the only
	 * new work is subtracting the cumulative counters so the step carries
	 * ITS usage rather than the turn's running total, which is the number a
	 * caller asking "what did this step cost" actually wants.
	 */
	/**
	 * Take everything queued for this turn since the last turn.
	 *
	 * Both channels drain here. `inboundMessages` is the manager's queue —
	 * what `continueTask` and `queueMessage` push onto and nothing ever
	 * collected. `steering` is the host's, and it could only ride on a tool
	 * result, so guidance queued during a turn that called no tools stayed
	 * pending until the turn ended.
	 *
	 * Returns the count so a caller can decide whether a turn is owed. An
	 * empty drain must change nothing at all: a `continue` on nothing queued
	 * spends an iteration and a model call to say the same thing again.
	 */
	private rememberUserMessage(message: Message, arriving = true): void {
		if (!isOperatorUserMessage(message)) return
		this.latestUserMessage = message
		// The hook field alone does not reach the model. Preserve arrivals in
		// the compaction state too, before their original message or attached
		// tool result can be shed. Initial history was already extracted at seed.
		const manager = this.ctx.workingStateManager
		if (arriving && manager) {
			extractFromUserMessage(manager, message.content, !manager.getState().task)
		}
	}

	private deliverInbound(): number {
		const queued = this.ctx.inboundMessages?.() ?? []
		for (const message of queued) {
			this.ctx.recorder.pushMessage(message)
			this.rememberUserMessage(message)
		}

		// The steering channel's remainder. `attachSteering` already took
		// what it could carry on a tool result; anything still pending is
		// guidance from a turn that had no result to attach it to.
		const stranded = this.ctx.steering?.drain()
		if (stranded) {
			const message = createRuntimeContextMessage(formatSteeringNote(stranded), 'steering')
			this.ctx.recorder.pushMessage(message)
			this.rememberUserMessage(createRuntimeContextMessage(stranded, 'steering'))
		}

		return queued.length + (stranded ? 1 : 0)
	}

	private recordStep(input: {
		stepNumber: number
		model: string
		servedBy?: StepProvenance
		messageId?: MessageId
		/**
		 * The turn's response. Absent only when the iteration failed before
		 * the provider produced one — see `unfinished`.
		 */
		response?: ChatCompletionResponse
		toolResults: readonly ToolCallOutcome[]
		toolExecutionMs: number
		startedAt: number
		usageBefore: TokenUsage
		costBefore: CostInfo
		/**
		 * Set only by the `catch`, for an iteration that did not finish.
		 *
		 * The same writer builds both records on purpose: a failed turn's
		 * step is a `StepResult` like any other, so a caller reconstructing
		 * cost or history sorts them together instead of discovering that
		 * failures live somewhere else.
		 */
		unfinished?: { finishReason: 'error' | 'cancelled'; failure?: StepFailure }
	}): void {
		const { recorder } = this.ctx
		const toolCalls = input.response?.message.toolCalls ?? []
		const byId = new Map(input.toolResults.map((r) => [r.toolCallId, r]))

		const step: StepResult = {
			stepNumber: input.stepNumber,
			model: input.model,
			...(input.servedBy ? { servedBy: input.servedBy } : {}),
			...(input.messageId ? { messageId: input.messageId } : {}),
			content: input.response?.message.content ?? null,
			toolCalls,
			// Ordered by the tool CALLS, not by completion, so the record
			// matches what the model asked for.
			//
			// On an unfinished step the calls with no outcome are DROPPED
			// rather than filled with `{output: '', isError: false}`. That
			// filler is a reading of "the batch was refused" on the success
			// path, where every call in a batch shares one verdict; under a
			// step that says `error` it would say a tool ran and returned
			// nothing successfully, which is the same lie one level down as
			// the missing step itself.
			toolResults: toolCalls.flatMap((tc) => {
				const outcome = byId.get(tc.id)
				if (input.unfinished && !outcome) return []
				return [
					{
						toolCallId: tc.id,
						toolName: tc.function.name,
						output: outcome?.output ?? '',
						isError: outcome?.isError ?? false,
						durationMs: 0,
					},
				]
			}),
			// The turn's own verdict where there is one. A step that ended in
			// the catch has none — no provider reported `error` or
			// `cancelled` — so `unfinished` wins even when a response had
			// already arrived: a turn that answered and then threw during
			// tool execution did not end in `tool_calls`.
			finishReason: input.unfinished?.finishReason ?? input.response?.finishReason ?? 'error',
			...(input.unfinished?.failure ? { failure: input.unfinished.failure } : {}),
			usage: subtractUsage(recorder.tokenUsage, input.usageBefore),
			costDelta: {
				...recorder.costInfo,
				totalCost: round6(recorder.costInfo.totalCost - input.costBefore.totalCost),
			},
			startedAt: input.startedAt,
			durationMs: Date.now() - input.startedAt,
			toolExecutionMs: input.toolExecutionMs,
		}

		this.steps.push(step)

		if (!input.unfinished) {
			this.ctx.onStepFinish?.(step)
			return
		}

		// Nothing here is allowed to throw over the failure that is already
		// unwinding — the same rule `settleCancelledTurn` states for the
		// cancellation path. A host callback that throws while being told a
		// turn failed would REPLACE the reason the turn failed, so the turn
		// would report the observer's bug and lose the original.
		try {
			this.ctx.onStepFinish?.(step)
		} catch (err) {
			this.ctx.log.warn('onStepFinish threw while recording a failed step', {
				[NAMZU.TURN_ID]: recorder.turnId,
				'namzu.runtime.step': input.stepNumber,
				'exception.message': toErrorMessage(err),
			})
		}
	}

	/** Turns spent asking the model again for a valid structured output. */
	private structuredOutputAttempts = 0
	private nativeStructuredAttempts = 0
	private structuredOutputDone = false
	private pendingStructuredOutput: unknown
	private structuredReviewAttempts = 0

	private structuredOutputRetryLimit(): number {
		return this.ctx.structuredOutput?.maxRetries ?? DEFAULT_STRUCTURED_OUTPUT_RETRIES
	}

	/** True while a structured output was demanded and has not arrived. */
	private needsStructuredOutput(): boolean {
		return this.ctx.structuredOutput !== undefined && !this.structuredOutputDone
	}

	/**
	 * The answer a terminal tool produced, or `undefined` to keep looping.
	 *
	 * Deliberately narrow. A terminal call decides the turn only when it is
	 * the ONLY call the model made in that turn: a model that asked for
	 * other work meant to see those results, and settling here would throw
	 * away answers it requested. Same for a failed terminal call — an
	 * error is not an answer, and the model is the one that should read
	 * it. Both cases fall through to the ordinary path, and both say so in
	 * the log rather than quietly costing the relay the flag was set to
	 * avoid.
	 */
	private terminalToolOutput(
		results: readonly ToolCallOutcome[],
		response: ChatCompletionResponse,
	): ToolCallOutcome | undefined {
		if (this.ctx.structuredOutput?.mode === 'native') return undefined
		const terminal = results.filter((r) => this.ctx.tools.get(r.toolName)?.terminal === true)
		if (terminal.length === 0) return undefined

		const callCount = response.message.toolCalls?.length ?? 0
		if (callCount > 1) {
			this.ctx.log.info('Terminal tool shared its turn — relaying instead of settling', {
				[NAMZU.TURN_ID]: this.ctx.recorder.turnId,
				[GENAI.TOOL_NAME]: terminal[0]?.toolName,
				'namzu.runtime.calls_in_turn': callCount,
			})
			return undefined
		}

		const hit = terminal[0]
		if (!hit || hit.isError) {
			this.ctx.log.info('Terminal tool failed — returning the error to the model', {
				[NAMZU.TURN_ID]: this.ctx.recorder.turnId,
				[GENAI.TOOL_NAME]: hit?.toolName,
			})
			return undefined
		}
		return hit
	}

	/**
	 * Record the structured output if this batch produced one.
	 *
	 * The tool validates against the Zod schema before its `execute` runs,
	 * so reaching here successfully means the value is already valid — a
	 * failed parse comes back as an error result and simply does not
	 * satisfy the demand, which sends the loop round again.
	 *
	 * Narrow in the same way {@link terminalToolOutput} is, for its stated
	 * reason and one that is sharper here. The neighbour refuses a shared
	 * turn because "a model that asked for other work meant to see those
	 * results". That applies unchanged. But the batch has ALREADY executed
	 * by the time this runs — `runToolReview` settles it, side effects
	 * included, before either of these is consulted — so settling here is
	 * worse than discarding an answer the model wanted: the work happened,
	 * its results went into the transcript, and the turn ended before any
	 * model turn could read them. Nothing consumed what was spent, and
	 * nothing said so.
	 *
	 * Sharper, too, because of WHEN this value was produced. The model
	 * emitted its final answer in the same turn as a request for
	 * information it had not yet received — it would not have asked
	 * otherwise — so the answer is under-informed on the model's own
	 * account, and settling ships it as final.
	 *
	 * So: relay, do not settle. The results are already in the transcript,
	 * the demand is still unsatisfied, and the next turn produces the
	 * answer with them in hand. Refusing to EXECUTE the batch was the other
	 * candidate and is wrong — the defect is not that the tools ran, it is
	 * that nobody read them, and denying a model work it asked for to
	 * protect an answer it has not finished forming gives up a real
	 * capability for nothing. The price is one extra turn when the paired
	 * call was a pure side effect whose result the model did not need;
	 * that is the price `terminalToolOutput` already pays, and a model
	 * avoids it by not pairing.
	 *
	 * NOT charged to `maxRetries`. That budget bounds a model that cannot
	 * satisfy the SCHEMA, and this one did. A turn reading two files a turn
	 * while optimistically attaching its answer is making progress, and it
	 * must not die reported as `structured_output_failed` — a failure that
	 * did not happen. `maxIterations` is the bound for a model that keeps
	 * doing work, and it is the bound the neighbour relies on for the
	 * identical pathology.
	 */
	private async captureStructuredOutput(
		results: readonly ToolCallOutcome[],
		response: ChatCompletionResponse,
		reviewRequest: ReviewRequest,
		model: string,
	): Promise<'absent' | 'accepted' | 'retry' | 'exhausted' | 'cancelled'> {
		if (!this.needsStructuredOutput() || this.ctx.structuredOutput?.mode === 'native')
			return 'absent'
		const hit = results.find((r) => r.toolName === STRUCTURED_OUTPUT_TOOL_NAME && !r.isError)
		if (!hit) return 'absent'

		const callCount = response.message.toolCalls?.length ?? 0
		if (callCount > 1) {
			this.ctx.log.info('Structured output shared its turn — relaying instead of settling', {
				[NAMZU.TURN_ID]: this.ctx.recorder.turnId,
				'namzu.runtime.calls_in_turn': callCount,
			})
			return 'absent'
		}

		let parsed: unknown
		try {
			parsed = JSON.parse(hit.output)
		} catch {
			// The tool serializes its own validated input, so this is
			// unreachable in practice; keep the raw text rather than losing it.
			if (this.ctx.structuredOutput?.review)
				throw new Error(
					'Structured review requires an intact JSON tool result; check tool-output limits and result transformations',
				)
			parsed = hit.output
		}
		return this.reviewStructuredOutput(parsed, reviewRequest, model)
	}

	private publishStructuredOutput(): void {
		this.ctx.recorder.setStructuredOutput(this.pendingStructuredOutput)
		this.structuredOutputDone = true
	}

	private async reviewStructuredOutput(
		parsed: unknown,
		reviewRequest: ReviewRequest,
		model: string,
	): Promise<'accepted' | 'retry' | 'exhausted' | 'cancelled'> {
		if (this.ctx.abortController.signal.aborted) return 'cancelled'
		const reviewer = this.ctx.structuredOutput?.review
		if (reviewer) {
			const signal = this.ctx.abortController.signal
			if (signal.aborted) return 'cancelled'
			let onAbort: () => void = () => {}
			const aborted = new Promise<never>((_resolve, reject) => {
				onAbort = () => reject(signal.reason ?? new Error('Structured review cancelled'))
				signal.addEventListener('abort', onAbort, { once: true })
			})
			let verdict: AnswerReview
			const inference = createCallbackInference(this.ctx, model, 'review')
			try {
				verdict = await Promise.race([
					Promise.resolve().then(() => {
						signal.throwIfAborted()
						return reviewer(structuredClone(parsed), {
							sessionId: this.ctx.recorder.sessionId,
							turnId: this.ctx.recorder.turnId,
							iteration: this.ctx.recorder.currentIteration,
							signal,
							messages: this.ctx.recorder.messages,
							...reviewRequest,
							generateText: inference.generateText,
						})
					}),
					aborted,
				])
			} catch (error) {
				if (signal.aborted) return 'cancelled'
				throw error
			} finally {
				inference.close()
				signal.removeEventListener('abort', onAbort)
			}
			if (signal.aborted) return 'cancelled'
			if (!verdict || typeof verdict.accept !== 'boolean')
				throw new Error('Structured reviewer returned an invalid verdict')
			if (!verdict.accept) {
				if (typeof verdict.feedback !== 'string' || verdict.feedback.trim().length === 0)
					throw new Error('Structured reviewer rejection requires feedback')
				this.structuredReviewAttempts++
				this.ctx.recorder.pushMessage(
					createRuntimeContextMessage(verdict.feedback, 'answer-review'),
				)
				// Persist both the feedback and its consumed allowance before the
				// next request. This checkpoint is not an approval/park boundary.
				const checkpoint = await this.ctx.checkpointMgr.create(
					this.ctx.recorder,
					this.ctx.recorder.currentIteration,
				)
				await this.ctx.emitEvent({
					type: 'checkpoint_created',
					turnId: this.ctx.recorder.turnId,
					checkpointId: checkpoint.id,
					iteration: this.ctx.recorder.currentIteration,
				})
				if (signal.aborted) return 'cancelled'
				return this.structuredReviewAttempts >
					(this.ctx.structuredOutput?.maxReviews ?? DEFAULT_ANSWER_REVIEW_LIMIT)
					? 'exhausted'
					: 'retry'
			}
		}
		this.pendingStructuredOutput = parsed
		return 'accepted'
	}

	/** A reviewer failure aborts settlement; only an explicit rejection requests correction. */
	private async reviewAnswer(
		answer: string,
		reviewRequest: ReviewRequest,
		model: string,
	): Promise<AnswerReview | undefined> {
		const reviewer = this.ctx.reviewAnswer
		const signal = this.ctx.abortController.signal
		if (!reviewer || signal.aborted) return undefined
		let onAbort = () => {}
		const aborted = new Promise<never>((_resolve, reject) => {
			onAbort = () => reject(signal.reason ?? new Error('Answer review cancelled'))
			signal.addEventListener('abort', onAbort, { once: true })
		})
		const inference = createCallbackInference(this.ctx, model, 'review')
		try {
			const verdict = await Promise.race([
				Promise.resolve().then(() => {
					signal.throwIfAborted()
					return reviewer(answer, {
						sessionId: this.ctx.recorder.sessionId,
						turnId: this.ctx.recorder.turnId,
						iteration: this.ctx.recorder.currentIteration,
						signal,
						messages: this.ctx.recorder.messages,
						...reviewRequest,
						generateText: inference.generateText,
					})
				}),
				aborted,
			])
			if (signal.aborted) return undefined
			if (!verdict || typeof verdict.accept !== 'boolean')
				throw new Error('Answer reviewer returned an invalid verdict')
			if (!verdict.accept && (typeof verdict.feedback !== 'string' || !verdict.feedback.trim()))
				throw new Error('Answer reviewer rejection requires feedback')
			return verdict
		} catch (error) {
			if (signal.aborted) return undefined
			throw new AnswerReviewFailure(error)
		} finally {
			inference.close()
			signal.removeEventListener('abort', onAbort)
		}
	}

	/** Evaluate the caller's halt predicate, if there is one. */
	private async shouldStop(): Promise<boolean> {
		const stopWhen = this.ctx.stopWhen
		const latestStep = this.steps.at(-1)
		if (!stopWhen || !latestStep) return false
		try {
			return await stopWhen({
				steps: this.steps,
				latestStep,
				totalUsage: this.ctx.recorder.tokenUsage,
				totalCost: this.ctx.recorder.costInfo,
			})
		} catch (err) {
			// A throwing predicate must not kill a turn that is otherwise
			// healthy; failing open keeps the existing budgets in charge.
			this.ctx.log.error('Stop condition threw — continuing the turn', {
				[NAMZU.TURN_ID]: this.ctx.recorder.turnId,
				'exception.message': toErrorMessage(err),
			})
			return false
		}
	}

	private hasRunningAgentTasks(): boolean {
		if (!this.ctx.taskGateway) return false
		return this.ctx.taskGateway
			.listTasks()
			.some((t) => t.state !== 'completed' && t.state !== 'failed' && t.state !== 'canceled')
	}

	private async acceptProviderRejectedImage(identity: RequestImageIdentity): Promise<void> {
		const repaired = markProviderRejectedImage(this.ctx.recorder.messages, identity)
		if (repaired.count === 0) {
			throw new Error('Provider-rejected image recovery could not find its durable source image')
		}
		this.ctx.recorder.replaceMessages(repaired.messages)
		await this.ctx.emitEvent?.({
			type: 'message_history_repaired',
			turnId: this.ctx.recorder.turnId,
			source: 'provider-rejected-image',
			duplicateToolResultsRemoved: 0,
			orphanedToolResultsRemoved: 0,
			syntheticToolResultsInserted: 0,
			providerRejectedImagesSuppressed: repaired.count,
		})
	}

	private async requestFinalResponse(
		model: string,
		reason: StopReason,
	): Promise<StopReason | undefined> {
		if (this.ctx.structuredOutput?.mode === 'native') return 'structured_output_failed'

		const lastAssistant = [...this.ctx.recorder.messages]
			.reverse()
			.find((m) => m.role === 'assistant')

		const hasResult =
			lastAssistant?.content !== null &&
			lastAssistant?.content !== undefined &&
			lastAssistant.content.length > 0

		if (hasResult) return
		// An empty completion may itself have exhausted the turn. This fallback
		// is another billed request, so it must pass the same hard limits as
		// the next normal iteration and preserve their unfinished stop reason.
		const guardResult = this.ctx.guard.beforeIteration(
			this.ctx.recorder,
			this.ctx.abortController.signal,
		)
		if (guardResult.shouldStop) return guardResult.stopReason

		this.ctx.log.info('Requesting final response after empty completion', {
			'namzu.runtime.reason': reason,
		})

		try {
			// The working-memory slot leaves the system run exactly as it does
			// for a normal step, and for the same cache reason.
			const workingMemory = splitWorkingMemoryForRequest(this.ctx.recorder.messages)
			const finalHistory = [
				...workingMemory.history,
				...(workingMemory.context ? [workingMemory.context] : []),
			]
			const finalMessages = projectRequestRichContent(
				this.projectObservations(finalHistory),
				this.ctx.turnConfig.maxRequestRichContentBytes ?? DEFAULT_MAX_REQUEST_RICH_CONTENT_BYTES,
			)
			appendWorkContext(this.stepShaping(), finalMessages, this.steps.length + 1, { model })
			// The closing directive is the last thing the model reads: after
			// the working memory and any work context above.
			finalMessages.push(
				createRuntimeContextMessage(
					`[SYSTEM] Turn is ending due to ${reason}. ${CLOSING_RESPONSE_GUIDANCE}`,
					'limit-finalization',
				),
			)
			await this.reportUnsupportedToolResults(finalMessages)

			// Same cache discipline as the forced-final iteration: keep the
			// tools param identical to prior iterations (cache prefix intact,
			// no 400 on tool blocks in history) and forbid use via tool_choice.
			const finalTools = this.ctx.tools.toLLMTools(this.ctx.allowedTools)
			const finalEnforced = enforcedModelInputToolNames(this.ctx.tools, finalTools)
			const requestedMember = this.ctx.servingMember?.() ?? {
				index: 0,
				providerId: this.ctx.provider.id,
			}
			const requestedRoute: StepProvenance = {
				providerId: requestedMember.providerId,
				model: requestedMember.model ?? model,
				chainIndex: requestedMember.index,
			}
			const finalParams = {
				model,
				providerRoute: requestedRoute,
				messages: finalMessages,
				tools: finalTools.length > 0 ? finalTools : undefined,
				...(finalEnforced ? { enforceToolInputSchema: finalEnforced } : {}),
				toolChoice: finalTools.length > 0 ? 'none' : undefined,
				temperature: this.ctx.turnConfig.temperature,
				maxTokens: this.ctx.turnConfig.maxResponseTokens,
				cacheControl: { type: 'auto' },
				...(this.ctx.turnConfig.thinking ? { thinking: this.ctx.turnConfig.thinking } : {}),
				// This turn is a hand-maintained duplicate of the one above, which
				// is exactly the shape a field goes missing from — so it is tested
				// separately rather than assumed to have been kept in step.
				...(this.ctx.turnConfig.effort ? { effort: this.ctx.turnConfig.effort } : {}),
				// Cancellable too: a Stop during the closing summary must not
				// stream to completion.
				signal: this.ctx.abortController.signal,
			} satisfies import('../../../types/provider/index.js').ChatCompletionParams
			const response = await collectChatCompletion(
				streamWithProviderRejectedImageRecovery(this.ctx.provider, finalParams, (identity) =>
					this.acceptProviderRejectedImage(identity),
				),
			)

			const servingMember = this.ctx.servingMember?.() ?? requestedMember
			const servedBy: StepProvenance = {
				providerId: servingMember.providerId,
				model: servingMember.model ?? model,
				chainIndex: servingMember.index,
			}
			this.ctx.recorder.accumulateUsage(response.usage, {
				providerId: servedBy.providerId,
				model: servedBy.model,
			})

			const assistantMsg = createAssistantMessage(
				response.message.content,
				undefined,
				response.message.reasoning,
				response.message.citations,
				{
					type: 'model',
					...servedBy,
					...(response.message.replayState !== undefined
						? { replayState: response.message.replayState }
						: {}),
				},
				response.message.textParts,
			)
			this.ctx.recorder.pushMessage(assistantMsg)

			const finalMessageId = generateMessageId()
			await this.ctx.emitEvent({
				type: 'message_started',
				turnId: this.ctx.recorder.turnId,
				iteration: this.ctx.recorder.currentIteration,
				messageId: finalMessageId,
			})
			await this.ctx.emitEvent({
				type: 'message_completed',
				turnId: this.ctx.recorder.turnId,
				iteration: this.ctx.recorder.currentIteration,
				messageId: finalMessageId,
				stopReason: 'forced_finalize',
				usage: response.usage,
				content: response.message.content ?? undefined,
				...(response.message.textParts ? { textParts: response.message.textParts } : {}),
			})
		} catch (err) {
			this.ctx.log.error('Failed to get final response', {
				'exception.message': toErrorMessage(err),
			})
		}
		return undefined
	}
}

/**
 * Fold whatever ended an iteration into the record a reader gets.
 *
 * Classified through `classifyProviderError` — the same call the catch
 * already makes to decide whether compaction relief applies — so the step's
 * verdict and the loop's own decision cannot drift apart. It also handles a
 * failure that is not a provider failure at all: the code set's `unknown`
 * means "unclassifiable", which is the true answer for a plugin hook that
 * threw and is left saying so rather than dressed up as something specific.
 */
function describeStepFailure(err: unknown, providerId: string): StepFailure {
	if (err instanceof AnswerReviewFailure)
		return { message: err.message, code: 'unknown', retryable: false }
	const classified = classifyProviderError(err, providerId)
	return {
		message: toErrorMessage(err),
		code: classified.code,
		...(classified.status !== undefined ? { status: classified.status } : {}),
		retryable: classified.retryable,
	}
}

/** Per-step usage: the delta between two cumulative snapshots. */
function subtractUsage(after: TokenUsage, before: TokenUsage): TokenUsage {
	return {
		promptTokens: after.promptTokens - before.promptTokens,
		completionTokens: after.completionTokens - before.completionTokens,
		totalTokens: after.totalTokens - before.totalTokens,
		cachedTokens: (after.cachedTokens ?? 0) - (before.cachedTokens ?? 0),
		cacheWriteTokens: (after.cacheWriteTokens ?? 0) - (before.cacheWriteTokens ?? 0),
	}
}

/** Cost deltas are subtractions of floats; keep them presentable. */
function round6(n: number): number {
	return Math.round(n * 1e6) / 1e6
}

/**
 * Which of the tools going out on this request have a closed model schema.
 *
 * A driver reads `enforceToolInputSchema` to decide which tool schemas to
 * constrain generation against. Nothing populated it, so every driver that
 * consumed it — three of them — was reading a permanently undefined field
 * and `enforceModelInput: true` on a tool meant nothing end to end.
 *
 * Computed per request rather than once, because the allowed set changes:
 * a deferred tool activated mid-run has to start being enforced from the
 * next call, not the next process.
 */
function enforcedModelInputToolNames(
	registry: ToolRegistryContract,
	tools: readonly LLMToolSchema[],
): readonly string[] | undefined {
	const names = tools
		.map((tool) => tool.function.name)
		.filter((name) => registry.get(name)?.enforceModelInput === true)
	return names.length > 0 ? names : undefined
}