package com.correx.core.kernel.orchestration

import com.correx.core.approvals.ApprovalOutcome
import com.correx.core.approvals.ApprovalStatus
import com.correx.core.approvals.Tier
import com.correx.core.approvals.DefaultApprovalRepository
import com.correx.core.approvals.domain.ApprovalEngine
import com.correx.core.approvals.isAtMost
import com.correx.core.approvals.model.ApprovalContext
import com.correx.core.approvals.model.ApprovalDecision
import com.correx.core.approvals.model.ApprovalScopeIdentity
import com.correx.core.approvals.model.DomainApprovalRequest
import com.correx.core.artifacts.ArtifactState
import com.correx.core.artifacts.kind.ArtifactKindRegistry
import com.correx.core.artifacts.kind.JsonSchema
import com.correx.core.artifacts.kind.FileWrittenArtifact
import com.correx.core.artifacts.kind.ProcessResultArtifact
import com.correx.core.artifacts.kind.TypedArtifactSlot
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.context.builder.ContextPackBuilder
import com.correx.core.context.model.ContextEntry
import com.correx.core.context.model.ContextLayer
import com.correx.core.context.model.ContextPack
import com.correx.core.context.model.TokenBudget
import com.correx.core.context.model.EntryRole
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
import com.correx.core.events.events.RefinementIterationEvent
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.ArtifactContentStoredEvent
import com.correx.core.events.events.ArtifactCreatedEvent
import com.correx.core.events.events.BriefEchoMismatchEvent
import com.correx.core.events.events.BriefGroundingCheckedEvent
import com.correx.core.events.events.ContextTruncatedEvent
import com.correx.core.events.events.ClarificationAnswer
import com.correx.core.events.events.ClarificationAnsweredEvent
import com.correx.core.events.events.ClarificationQuestions
import com.correx.core.events.events.ClarificationRequestedEvent
import com.correx.core.events.events.GroundingReference
import com.correx.core.events.events.ToolCallAssessedEvent
import com.correx.core.events.events.ArtifactValidatedEvent
import com.correx.core.events.events.ArtifactValidatingEvent
import com.correx.core.events.events.EventMetadata
import com.correx.core.events.events.EventPayload
import com.correx.core.events.events.InferenceCompletedEvent
import com.correx.core.events.events.RepoKnowledgeRetrievedEvent
import com.correx.core.events.events.RepoMapComputedEvent
import com.correx.core.events.events.InferenceFailedEvent
import com.correx.core.events.events.InferenceStartedEvent
import com.correx.core.events.events.InferenceTimeoutEvent
import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.RiskAssessedEvent
import com.correx.core.events.events.ExecutionPlanLockedEvent
import com.correx.core.events.events.FileWrittenEvent
import com.correx.core.events.events.StageCheckpointFailedEvent
import com.correx.core.events.events.StageCheckpointPassedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.ToolExecutionCompletedEvent
import com.correx.core.events.events.ToolExecutionFailedEvent
import com.correx.core.events.events.ToolExecutionRejectedEvent
import com.correx.core.events.events.ToolInvocationRequestedEvent
import com.correx.core.events.events.ToolReceipt
import com.correx.core.events.events.ToolRequest
import com.correx.core.events.risk.RiskAction
import com.correx.core.events.risk.RiskSummary
import com.correx.core.toolintent.ToolCallAssessmentInput
import com.correx.core.toolintent.ToolCallAssessor
import com.correx.core.toolintent.WorkspacePolicy
import com.correx.core.toolintent.WorldProbe
import com.correx.core.toolintent.toAssessedIssues
import com.correx.core.toolintent.toRiskSummary
import com.correx.core.events.events.TransitionExecutedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.ApprovalDecisionId
import com.correx.core.events.types.ApprovalRequestId
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.ClarificationRequestId
import com.correx.core.events.types.ContextEntryId
import com.correx.core.events.types.ContextPackId
import com.correx.core.events.types.EventId
import com.correx.core.events.types.InferenceRequestId
import com.correx.core.events.types.RiskSummaryId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.events.types.ToolInvocationId
import com.correx.core.events.types.ValidationReportId
import com.correx.core.inference.FinishReason
import com.correx.core.inference.InferenceRepository
import com.correx.core.inference.InferenceRequest
import com.correx.core.inference.InferenceResponse
import com.correx.core.inference.InferenceRouter
import com.correx.core.inference.PromptRenderer
import com.correx.core.inference.ResponseFormat
import com.correx.core.inference.Tokenizer
import com.correx.core.inference.ToolCallRequest
import com.correx.core.inference.ToolDefinition
import com.correx.core.sessions.ApprovalMode
import com.correx.core.inference.ToolFunction
import com.correx.core.kernel.execution.WorkflowResult
import com.correx.core.risk.RiskAssessor
import com.correx.core.risk.RiskContext
import com.correx.core.risk.toApprovalTier
import com.correx.core.sessions.Session
import com.correx.core.tools.compression.ToolOutputContext
import com.correx.core.tools.contract.FileAffectingTool
import com.correx.core.tools.contract.Tool
import com.correx.core.tools.contract.ToolExecutor
import com.correx.core.tools.contract.ToolResult
import com.correx.core.tools.registry.ToolRegistry
import com.correx.core.transitions.evaluation.EvaluationContext
import com.correx.core.transitions.evaluation.PromptResolver
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.transitions.graph.StageConfig
import com.correx.core.transitions.graph.WorkflowGraph
import com.correx.core.transitions.resolution.TransitionDecision
import com.correx.core.transitions.resolution.TransitionResolver
import com.correx.core.validation.model.ValidationContext
import com.correx.core.validation.pipeline.ValidationOutcome
import com.correx.core.validation.pipeline.ValidationPipeline
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.withContext
import kotlinx.coroutines.withTimeout
import kotlinx.datetime.Clock
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonObject
import com.correx.core.journal.DecisionJournalRenderer
import com.correx.core.journal.DefaultDecisionJournalRepository
import com.correx.core.kernel.orchestration.subagent.SubagentRunner
import org.slf4j.LoggerFactory
import java.nio.file.Files
import java.nio.file.Paths
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
import kotlin.coroutines.cancellation.CancellationException

private const val MAX_TOOL_ROUNDS = 10
private const val STAGE_COMPLETE_TOOL = "stage_complete"
private const val OUTPUT_SUMMARY_LIMIT = 500
private const val REPO_MAP_INJECT_TOP_K = 30

// HTTP statuses that are transient despite being 4xx (F-002 retry classification).
private const val HTTP_TIMEOUT = 408
private const val HTTP_TOO_MANY_REQUESTS = 429

// Cap on clarification rounds per stage, so a stage that keeps re-asking eventually proceeds.
private const val MAX_CLARIFICATION_ROUNDS = 3

@SuppressWarnings(
    "ForbiddenComment",
    "UnusedParameter",
    "ReturnCount",
    "TooGenericExceptionCaught",
    "TooManyFunctions",
    "NestedBlockDepth",
    "LongMethod",
)
abstract class SessionOrchestrator(
    repositories: OrchestratorRepositories,
    engines: OrchestratorEngines,
    protected val artifactStore: ArtifactStore,
    private val decisionJournalRepository: DefaultDecisionJournalRepository,
    private val decisionJournalRenderer: DecisionJournalRenderer = DecisionJournalRenderer(),
    private val artifactKindRegistry: ArtifactKindRegistry? = null,
    private val repoKnowledgeRetriever: RepoKnowledgeRetriever? = null,
) {
    private val log = LoggerFactory.getLogger(this::class.java)
    private val eventStore: EventStore = repositories.eventStore
    private val transitionResolver: TransitionResolver = engines.transitionResolver
    private val contextPackBuilder: ContextPackBuilder = engines.contextPackBuilder
    private val inferenceRouter: InferenceRouter = engines.inferenceRouter
    private val validationPipeline: ValidationPipeline = engines.validationPipeline
    private val riskAssessor: RiskAssessor = engines.riskAssessor
    private val promptResolver: PromptResolver = engines.promptResolver
    private val toolExecutor: ToolExecutor? = engines.toolExecutor
    private val toolRegistry: ToolRegistry? = engines.toolRegistry
    private val toolCallAssessor: ToolCallAssessor? = engines.toolCallAssessor
    private val workspacePolicy: WorkspacePolicy? = engines.workspacePolicy
    private val worldProbe: WorldProbe = engines.worldProbe
    private val workspaceToolRegistryProvider: WorkspaceToolRegistryProvider? = engines.workspaceToolRegistryProvider
    private val inferenceRepository: InferenceRepository = repositories.inferenceRepository
    internal val orchestrationRepository: OrchestrationRepository = repositories.orchestrationRepository
    protected open val tokenizer: Tokenizer? = null
    private val approvalEngine: ApprovalEngine = engines.approvalEngine
    private val approvalRepository: DefaultApprovalRepository = repositories.approvalRepository
    internal abstract val cancellations: ConcurrentHashMap<SessionId, AtomicBoolean>

    protected abstract val subagentRunner: SubagentRunner
    internal val pendingApprovals: ConcurrentHashMap<ApprovalRequestId, CompletableDeferred<ApprovalDecision>> =
        ConcurrentHashMap()

    // A stage whose produced artifact carries open questions parks its run on one of these until the
    // operator answers (submitClarification completes it). Separate from pendingApprovals so a
    // clarification round never consumes the failure-retry budget.
    internal val pendingClarifications:
        ConcurrentHashMap<ClarificationRequestId, CompletableDeferred<List<ClarificationAnswer>>> =
        ConcurrentHashMap()

    /** Cache mapping "sessionId:artifactId" to JSON-serialized artifact payload content.
     *  Populated when ProcessResult artifacts are stored on tool failure.
     *  Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */
    protected val artifactContentCache: ConcurrentHashMap<String, String> = ConcurrentHashMap()

    abstract suspend fun run(
        sessionId: SessionId,
        graph: WorkflowGraph,
        config: OrchestrationConfig,
    ): WorkflowResult

    abstract suspend fun cancel(sessionId: SessionId)

    // --- per-run effective registry + policy ---

    internal data class RunEffectives(
        val registry: ToolRegistry?,
        val executor: ToolExecutor?,
        val policy: WorkspacePolicy?,
    )

    internal fun effectivesFor(config: OrchestrationConfig): RunEffectives {
        val wsTools = config.workspace?.let { workspaceToolRegistryProvider?.forWorkspace(it) }
        val registry = wsTools?.registry ?: toolRegistry
        val executor = wsTools?.executor ?: toolExecutor
        val policy = config.workspace
            ?.let { WorkspacePolicy(it.workspaceRoot, it.privilegedLocations) }
            ?: workspacePolicy
        return RunEffectives(registry, executor, policy)
    }

    // --- stage execution ---

    @Suppress("CyclomaticComplexMethod")
    internal suspend fun executeStage(
        sessionId: SessionId,
        stageId: StageId,
        graph: WorkflowGraph,
        session: Session,
        config: OrchestrationConfig,
        effectives: RunEffectives,
    ): StageExecutionResult {
        val stageConfig = requireNotNull(graph.stages[stageId]) {
            "Stage '${stageId.value}' not declared in workflow graph"
        }
        val systemPrompt = stageConfig.metadata["systemPrompt"]
            ?.let { path ->
                runCatching { promptResolver.resolve(path) }
                    .onFailure {
                        log.error(error(stageId, path, it))
                    }
                    .getOrNull()
            }
            ?.takeIf { it.isNotBlank() }
            ?.let { text ->
                listOf(
                    ContextEntry(
                        id = ContextEntryId(UUID.randomUUID().toString()),
                        layer = ContextLayer.L0,
                        content = text,
                        sourceType = "systemPrompt",
                        sourceId = stageId.value,
                        tokenEstimate = estimateTokens(text),
                        role = EntryRole.SYSTEM,
                    ),
                )
            } ?: config.defaultSystemPromptPath
            ?.let { path ->
                runCatching { promptResolver.resolve(path) }
                    .onFailure {
                        log.error(error(stageId, path, it))
                    }
                    .getOrNull()
            }
            ?.takeIf { it.isNotBlank() }
            ?.let { text ->
                listOf(
                    ContextEntry(
                        id = ContextEntryId(UUID.randomUUID().toString()),
                        layer = ContextLayer.L0,
                        content = text,
                        sourceType = "systemPrompt",
                        sourceId = stageId.value,
                        tokenEstimate = estimateTokens(text),
                        role = EntryRole.SYSTEM,
                    ),
                )
            } ?: emptyList()
        val promptEntries = stageConfig.metadata["promptInline"]
            ?.takeIf { it.isNotBlank() }
            ?.let { text ->
                listOf(
                    ContextEntry(
                        id = ContextEntryId(UUID.randomUUID().toString()),
                        layer = ContextLayer.L1,
                        content = text,
                        sourceType = "agentPrompt",
                        sourceId = stageId.value,
                        tokenEstimate = estimateTokens(text),
                        role = EntryRole.USER,
                    ),
                )
            }
            ?: stageConfig.metadata["prompt"]
                ?.let { path ->
                    val resolvedText = runCatching { promptResolver.resolve(path) }
                        .onFailure {
                            log.error(
                                "[SessionOrchestrator] stage=${stageId.value}: " +
                                        "failed to load prompt '$path': ${it.message}",
                            )
                        }
                        .getOrNull()
                    val text = resolvedText?.takeIf { it.isNotBlank() }
                        ?: error(
                            "[SessionOrchestrator] stage=${stageId.value}: " +
                                    "declared prompt '$path' could not be resolved",
                        )
                    listOf(
                        ContextEntry(
                            id = ContextEntryId(UUID.randomUUID().toString()),
                            layer = ContextLayer.L1,
                            content = text,
                            sourceType = "agentPrompt",
                            sourceId = stageId.value,
                            tokenEstimate = estimateTokens(text),
                            role = EntryRole.USER,
                        ),
                    )
                }
            ?: emptyList()

        val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted }
        require(llmEmittedSlots.size <= 1) {
            "Stage '${stageId.value}' declares ${llmEmittedSlots.size} LLM-emitted artifact kinds; " +
                    "only 0 or 1 is supported"
        }
        val responseFormat = llmEmittedSlots.firstOrNull()
            ?.let { ResponseFormat.Json(it.kind.deriveJsonSchema()) }
            ?: ResponseFormat.Text

        val schemaEntries = buildSchemaEntries(responseFormat, stageId)
        val steeringEntries = buildSteeringNoteEntries(sessionId)
        val clarificationEntries = buildClarificationAnswerEntries(sessionId)

        // Maintain the running transcript in true chronological order. Reading it back
        // from currentContext.layers (a Map grouped by layer) would scramble turn order
        // across rounds; instead we grow our own ordered list and let the builder restamp
        // ordinals from it each round.
        val journalState = decisionJournalRepository.getJournal(sessionId)
        val summaryText = journalState.summaryArtifactId?.let { id ->
            artifactStore.get(id)?.toString(Charsets.UTF_8)
                ?: run {
                    log.warn("Journal summary artifact {} not found in store — rendering without summary", id.value)
                    null
                }
        }
        val journalText = decisionJournalRenderer.render(journalState, summaryText)
        val journalEntries = if (journalText.isBlank()) emptyList() else listOf(
            ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L0,
                content = journalText,
                sourceType = "decisionJournal",
                sourceId = "decision-journal",
                tokenEstimate = journalText.length / 4,
                role = EntryRole.SYSTEM,
            ),
        )
        val repoMapEntries = buildContextualRepoEntries(sessionId, stageId, stageConfig)
        val sessionEvents = eventStore.read(sessionId)
        val criticIds = criticArtifactIds(sessionEvents, graph, stageId)
        val criticFrom = sessionEvents
            .mapNotNull { it.payload as? RefinementIterationEvent }
            .lastOrNull { it.cycleKey.endsWith("->${stageId.value}") }
            ?.cycleKey?.substringBefore("->")
        val needsEntries = buildNeedsArtifactEntries(sessionId, stageConfig, criticIds, criticFrom)
        val profileEntries = session.state.boundProfile?.about
            ?.takeIf { it.isNotBlank() }
            ?.let { about ->
                listOf(
                    ContextEntry(
                        id = ContextEntryId(UUID.randomUUID().toString()),
                        layer = ContextLayer.L0,
                        content = "## Operator profile\n$about",
                        sourceType = "operatorProfile",
                        sourceId = "operator-profile",
                        tokenEstimate = about.length / 4,
                        role = EntryRole.SYSTEM,
                    ),
                )
            } ?: emptyList()
        val projectProfileEntries = session.state.boundProjectProfile
            ?.let { listOf(buildProjectProfileEntry(it)) } ?: emptyList()
        val retryFeedbackEntries = buildRetryFeedbackEntry(sessionEvents, stageId)
            ?.let { listOf(it) } ?: emptyList()
        val vocabularyEntries = artifactKindRegistry
            ?.takeIf { stageConfig.metadata["injectArtifactKinds"] == "true" }
            ?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList()
        var accumulatedEntries =
            systemPrompt + profileEntries + projectProfileEntries + journalEntries + repoMapEntries +
                needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries +
                clarificationEntries + retryFeedbackEntries
        val contextPack = contextPackBuilder.build(
            id = ContextPackId(UUID.randomUUID().toString()),
            sessionId = sessionId,
            stageId = stageId,
            entries = accumulatedEntries,
            budget = TokenBudget(limit = stageConfig.tokenBudget),
        )
        emitContextTruncationIfNeeded(sessionId, stageId, contextPack)

        var currentContext = contextPack
        var inferenceResult = runInference(
            sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
        )
        var toolRounds = 0

        val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" }
        fun owesFileWrite(): Boolean = fileWrittenSlots.isNotEmpty() &&
            fileWrittenSlots.any { artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() }

        // Append a corrective tool-result and re-run inference (bounded by MAX_TOOL_ROUNDS).
        // Returns the new result rather than mutating inferenceResult, to preserve smart casts.
        suspend fun pushBack(nudge: String): InferenceResult {
            accumulatedEntries = accumulatedEntries + ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L2,
                sourceType = "toolResult",
                sourceId = UUID.randomUUID().toString(),
                content = nudge,
                tokenEstimate = estimateTokens(nudge),
                role = EntryRole.TOOL,
            )
            currentContext = contextPackBuilder.build(
                id = ContextPackId(UUID.randomUUID().toString()),
                sessionId = sessionId,
                stageId = stageId,
                entries = accumulatedEntries,
                budget = TokenBudget(limit = stageConfig.tokenBudget),
            )
            emitContextTruncationIfNeeded(sessionId, stageId, currentContext)
            toolRounds++
            return runInference(
                sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
            )
        }

        val writeNudge = "ERROR: you described the change but did not apply it. You MUST call the " +
            "file_edit or file_write tool to write the change to the file before finishing — do " +
            "not output the file contents as a message."

        while (
            inferenceResult is InferenceResult.Success &&
            toolRounds < MAX_TOOL_ROUNDS &&
            (inferenceResult.response.finishReason is FinishReason.ToolCall || owesFileWrite())
        ) {
            // Content turn (no tool call) but the stage still owes a file_written artifact: the
            // model emitted the change as prose instead of invoking the write tool. Nudge it to
            // actually write (F-018).
            if (inferenceResult.response.finishReason !is FinishReason.ToolCall) {
                inferenceResult = pushBack(writeNudge)
                continue
            }
            // stage_complete is a meta-tool: the LLM calls it to signal the stage's goal is met.
            // Skip tool dispatch — the loop exits and normal validation/transition follows
            // (emitToolArtifacts + verifyProduces run on the success path after the loop).
            if (inferenceResult.response.toolCalls.any { it.function.name == STAGE_COMPLETE_TOOL }) {
                // Premature-completion guard: don't accept stage_complete while the stage still
                // owes an artifact. For an LLM-emitted slot, nudge to emit the JSON (F-010); for a
                // file_written slot, nudge to invoke the write tool (F-018). Otherwise complete.
                val owedLlm = llmEmittedSlots.firstOrNull()?.takeIf {
                    inferenceResult.response.text.isBlank() &&
                        artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank()
                }
                when {
                    owedLlm != null -> inferenceResult = pushBack(
                        "ERROR: stage_complete is not allowed yet — you have not emitted the " +
                            "required '${owedLlm.name.value}' artifact. Respond with the artifact as " +
                            "a single JSON message (no tool call) matching the provided schema.",
                    )
                    owesFileWrite() -> inferenceResult = pushBack(writeNudge)
                    else -> break
                }
                continue
            }
            val toolEntries = dispatchToolCalls(sessionId, stageId,
                inferenceResult.response.toolCalls, stageConfig, effectives)
            val fatalEntry = toolEntries.firstOrNull { it.content.startsWith("FATAL:") }
            if (fatalEntry != null) {
                emitProcessResultEvents(sessionId, stageId, stageConfig)
                return StageExecutionResult.Failure(
                    fatalEntry.content.removePrefix("FATAL: "),
                    retryable = false,
                )
            }
            // Recoverable (ERROR:) tool failures are NOT terminal — they flow back into the
            // loop as tool-result context so the model can see the error and adapt (bounded by
            // MAX_TOOL_ROUNDS). Only FATAL: failures (handled above) abort the stage.
            accumulatedEntries = accumulatedEntries + toolEntries
            currentContext = contextPackBuilder.build(
                id = ContextPackId(UUID.randomUUID().toString()),
                sessionId = sessionId,
                stageId = stageId,
                entries = accumulatedEntries,
                budget = TokenBudget(limit = stageConfig.tokenBudget),
            )
            emitContextTruncationIfNeeded(sessionId, stageId, currentContext)
            inferenceResult = runInference(
                sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
            )
            toolRounds++
        }
        return when (inferenceResult) {
            is InferenceResult.Cancelled -> StageExecutionResult.Failure("CANCELLED", retryable = false)
            is InferenceResult.Failed ->
                StageExecutionResult.Failure(inferenceResult.reason, retryable = isRetryableFailure(inferenceResult.reason))
            is InferenceResult.Success -> {
                if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false)
                // Capture the LLM-emitted artifact's JSON so transition conditions
                // (e.g. ArtifactFieldEquals on a reviewer's verdict) can read it.
                llmEmittedSlots.firstOrNull()?.let { slot ->
                    artifactContentCache["${sessionId.value}:${slot.name.value}"] =
                        inferenceResult.response.text
                    inferenceResult.responseArtifactId?.let { hash ->
                        emit(
                            sessionId,
                            ArtifactContentStoredEvent(
                                artifactId = slot.name,
                                contentHash = hash,
                                sessionId = sessionId,
                                stageId = stageId,
                            ),
                        )
                    }
                }
                val validationCtx = ValidationContext(
                    graph = graph,
                    sessionState = session.state,
                    availableTools = effectives.registry?.all()?.map { it.name }?.toSet(),
                    producedArtifactContent = graph.stages.values.flatMap { it.produces }.mapNotNull { slot ->
                        artifactContentCache["${sessionId.value}:${slot.name.value}"]?.let { slot.name.value to it }
                    }.toMap(),
                )
                when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) {
                    is StageExecutionResult.Success -> {
                        emitToolArtifacts(sessionId, stageId, stageConfig)
                        emitLlmArtifacts(sessionId, stageId, stageConfig)
                        val produced = verifyProduces(sessionId, stageId, stageConfig)
                        emitStageCheckpoint(sessionId, stageId, stageConfig)
                        when (produced) {
                            is StageExecutionResult.Success ->
                                when (val grounded = groundBriefReferences(sessionId, stageId, stageConfig, effectives)) {
                                    is StageExecutionResult.Success ->
                                        checkBriefEcho(sessionId, stageId, stageConfig)
                                    is StageExecutionResult.Failure -> grounded
                                }
                            is StageExecutionResult.Failure -> produced
                        }
                    }

                    is StageExecutionResult.Failure -> outcome
                }
            }
        }
    }

    private fun error(
        stageId: StageId,
        path: String,
        throwable: Throwable,
    ): String = "[SessionOrchestrator] stage=${stageId.value}: " +
            "failed to load prompt '$path': ${throwable.message}"

    /**
     * Strip an outer markdown code fence (```` ```json `` / ``` ``` ````) when the whole response
     * is a single fenced block. Conservative: only acts when the trimmed text both starts and ends
     * with a fence, so prose responses that merely contain an inline block are left untouched.
     */
    private fun stripCodeFence(text: String): String {
        val trimmed = text.trim()
        if (!trimmed.startsWith("```") || !trimmed.endsWith("```")) return text
        val withoutClose = trimmed.removeSuffix("```").trimEnd()
        val firstNewline = withoutClose.indexOf('\n')
        if (firstNewline < 0) return text
        // Drop the opening fence line (``` optionally followed by a language tag).
        return withoutClose.substring(firstNewline + 1).trim()
    }

    @Suppress("CyclomaticComplexMethod")
    private suspend fun dispatchToolCalls(
        sessionId: SessionId,
        stageId: StageId,
        toolCalls: List<ToolCallRequest>,
        stageConfig: StageConfig,
        effectives: RunEffectives,
    ): List<ContextEntry> {
        val executor = effectives.executor ?: return emptyList()
        val processResultSlots = stageConfig.produces.filter { it.kind.id == "process_result" }
        val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" }
        return toolCalls.flatMap { toolCall ->
            val invocationId = ToolInvocationId(UUID.randomUUID().toString())
            val parameters = runCatching {
                val element = Json.parseToJsonElement(toolCall.function.arguments)
                val jsonObject = element as? kotlinx.serialization.json.JsonObject ?: return@runCatching emptyMap()
                jsonObject.entries.associate { (k, v) ->
                    k to ((v as? kotlinx.serialization.json.JsonPrimitive)?.content ?: v.toString()) as Any
                }
            }.getOrElse { emptyMap() }
            val request = ToolRequest(
                invocationId = invocationId,
                sessionId = sessionId,
                stageId = stageId,
                toolName = toolCall.function.name,
                parameters = parameters,
            )
            val tool = effectives.registry?.resolve(toolCall.function.name)
            val tier = tool?.tier ?: Tier.T2
            emit(
                sessionId,
                ToolInvocationRequestedEvent(
                    invocationId = invocationId,
                    sessionId = sessionId,
                    stageId = stageId,
                    toolName = toolCall.function.name,
                    tier = tier,
                    request = request,
                ),
            )
            if (toolCall.function.name != STAGE_COMPLETE_TOOL &&
                toolCall.function.name !in stageConfig.allowedTools
            ) {
                val denyReason = "tool '${toolCall.function.name}' is not permitted in stage '${stageId.value}'"
                emit(
                    sessionId,
                    ToolExecutionRejectedEvent(
                        invocationId = invocationId,
                        sessionId = sessionId,
                        toolName = toolCall.function.name,
                        tier = tier,
                        reason = denyReason,
                    ),
                )
                val sourceId = toolCall.id ?: invocationId.value
                return@flatMap listOf(
                    ContextEntry(
                        id = ContextEntryId(UUID.randomUUID().toString()),
                        layer = ContextLayer.L2,
                        sourceType = "assistantToolCall",
                        sourceId = sourceId,
                        content = Json.encodeToString(ToolCallRequest.serializer(), toolCall),
                        tokenEstimate = estimateTokens(toolCall.function.arguments),
                        role = EntryRole.ASSISTANT,
                    ),
                    ContextEntry(
                        id = ContextEntryId(UUID.randomUUID().toString()),
                        layer = ContextLayer.L2,
                        sourceType = "toolResult",
                        sourceId = sourceId,
                        content = "ERROR: $denyReason",
                        tokenEstimate = estimateTokens(denyReason),
                        role = EntryRole.TOOL,
                    ),
                )
            }
            val plane2Risk: RiskSummary? = runPlane2Assessment(
                sessionId, stageId, invocationId, toolCall.function.name, request, tool, effectives,
                stageConfig.writeManifest,
            )?.let { assessment ->
                if (assessment.recommendedAction == RiskAction.BLOCK) {
                    emit(
                        sessionId,
                        ToolExecutionRejectedEvent(
                            invocationId = invocationId,
                            sessionId = sessionId,
                            toolName = toolCall.function.name,
                            tier = tier,
                            reason = "blocked by tool-call policy",
                        ),
                    )
                    val sourceId = toolCall.id ?: invocationId.value
                    return@flatMap listOf(
                        ContextEntry(
                            id = ContextEntryId(UUID.randomUUID().toString()),
                            layer = ContextLayer.L2,
                            sourceType = "assistantToolCall",
                            sourceId = sourceId,
                            content = Json.encodeToString(ToolCallRequest.serializer(), toolCall),
                            tokenEstimate = estimateTokens(toolCall.function.arguments),
                            role = EntryRole.ASSISTANT,
                        ),
                        ContextEntry(
                            id = ContextEntryId(UUID.randomUUID().toString()),
                            layer = ContextLayer.L2,
                            sourceType = "toolResult",
                            sourceId = sourceId,
                            content = "BLOCKED: ${assessment.rationale.joinToString("; ")}",
                            tokenEstimate = estimateTokens(assessment.rationale.joinToString("; ")),
                            role = EntryRole.TOOL,
                        ),
                    )
                }
                assessment
            }
            val plane2Prompts = plane2Risk?.recommendedAction == RiskAction.PROMPT_USER
            // A steering note attached to a human approval is captured here and injected after the
            // tool result, so the same-stage loop re-infers with it and the model acts on the note.
            var approvalNote: String? = null
            if (tier.isAtMost(Tier.T1) && !plane2Prompts) {
                // no approval needed
            } else {
                val approvalState = approvalRepository.getApprovalState(sessionId)
                val activeGrants = approvalState.grants.values.toList()
                val approvalCtx = ApprovalContext(
                    identity = ApprovalScopeIdentity(sessionId, stageId, projectId = null),
                    mode = ApprovalMode.PROMPT,
                )
                val requestId = ApprovalRequestId(UUID.randomUUID().toString())
                val toolPreview = computeToolPreview(toolCall.function.name, parameters)
                val domainRequest = DomainApprovalRequest(
                    id = requestId,
                    tier = tier,
                    validationReportId = ValidationReportId(UUID.randomUUID().toString()),
                    riskSummaryId = null,
                    timestamp = Clock.System.now(),
                    toolName = toolCall.function.name,
                    preview = toolPreview ?: toolCall.function.arguments.take(200),
                )
                val engineDecision = approvalEngine.evaluate(
                    domainRequest, approvalCtx, activeGrants, Clock.System.now(),
                )
                if (engineDecision.state == ApprovalStatus.COMPLETED) {
                    emitDecisionResolved(sessionId, domainRequest, engineDecision)
                    if (!engineDecision.isApproved) {
                        val rejectReason = engineDecision.reason ?: "denied"
                        emit(sessionId, ToolExecutionRejectedEvent(
                            invocationId = invocationId,
                            sessionId = sessionId,
                            toolName = toolCall.function.name,
                            tier = tier,
                            reason = rejectReason,
                        ))
                        val sourceId = toolCall.id ?: invocationId.value
                        return@flatMap listOf(
                            ContextEntry(
                                id = ContextEntryId(UUID.randomUUID().toString()),
                                layer = ContextLayer.L2,
                                sourceType = "assistantToolCall",
                                sourceId = sourceId,
                                content = Json.encodeToString(ToolCallRequest.serializer(), toolCall),
                                tokenEstimate = estimateTokens(toolCall.function.arguments),
                                role = EntryRole.ASSISTANT,
                            ),
                            ContextEntry(
                                id = ContextEntryId(UUID.randomUUID().toString()),
                                layer = ContextLayer.L2,
                                sourceType = "toolResult",
                                sourceId = sourceId,
                                content = "ERROR: $rejectReason",
                                tokenEstimate = estimateTokens(rejectReason),
                                role = EntryRole.TOOL,
                            ),
                        )
                    }
                    // grant auto-approved — fall through to execute
                } else {
                    val deferred = CompletableDeferred<ApprovalDecision>()
                    pendingApprovals[requestId] = deferred
                    emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING"))
                    emit(
                        sessionId,
                        ApprovalRequestedEvent(
                            requestId = requestId,
                            tier = tier,
                            validationReportId = domainRequest.validationReportId,
                            riskSummaryId = null,
                            riskSummary = plane2Risk,
                            sessionId = sessionId,
                            stageId = stageId,
                            projectId = null,
                            toolName = toolCall.function.name,
                            preview = toolPreview ?: toolCall.function.arguments.take(200),
                        ),
                    )
                    val userDecision = try {
                        deferred.await()
                    } finally {
                        pendingApprovals.remove(requestId)
                    }
                    emitDecisionResolved(sessionId, domainRequest, userDecision)
                    if (!userDecision.isApproved) {
                        val rejectReason = userDecision.reason ?: "approval denied"
                        emit(sessionId, ToolExecutionRejectedEvent(
                            invocationId = invocationId,
                            sessionId = sessionId,
                            toolName = toolCall.function.name,
                            tier = tier,
                            reason = rejectReason,
                        ))
                        emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
                        val sourceId = toolCall.id ?: invocationId.value
                        return@flatMap listOf(
                            ContextEntry(
                                id = ContextEntryId(UUID.randomUUID().toString()),
                                layer = ContextLayer.L2,
                                sourceType = "assistantToolCall",
                                sourceId = sourceId,
                                content = Json.encodeToString(ToolCallRequest.serializer(), toolCall),
                                tokenEstimate = estimateTokens(toolCall.function.arguments),
                                role = EntryRole.ASSISTANT,
                            ),
                            ContextEntry(
                                id = ContextEntryId(UUID.randomUUID().toString()),
                                layer = ContextLayer.L2,
                                sourceType = "toolResult",
                                sourceId = sourceId,
                                content = "ERROR: $rejectReason",
                                tokenEstimate = estimateTokens(rejectReason),
                                role = EntryRole.TOOL,
                            ),
                        )
                    }
                    approvalNote = userDecision.userSteering?.text
                    emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
                }
            }
            val result = executor.execute(request)

            // Store ProcessResult artifact for every shell execution outcome
            for (slot in processResultSlots) {
                val processResult = when (result) {
                    is ToolResult.Success -> ProcessResultArtifact(
                        status = "success",
                        command = toolCall.function.arguments,
                        exitCode = result.exitCode,
                        stdout = result.output,
                        stderr = result.metadata["stderr"] ?: "",
                        durationMs = 0,
                    )
                    is ToolResult.Failure -> ProcessResultArtifact(
                        status = "failed",
                        command = toolCall.function.arguments,
                        exitCode = -1,
                        stdout = "",
                        stderr = result.reason,
                        durationMs = 0,
                    )
                }
                val jsonContent = Json.encodeToString(ProcessResultArtifact.serializer(), processResult)
                val storedHash = artifactStore.put(jsonContent.toByteArray())
                artifactContentCache["${sessionId.value}:${slot.name.value}"] = jsonContent
                emit(
                    sessionId,
                    ArtifactContentStoredEvent(
                        artifactId = slot.name,
                        contentHash = storedHash,
                        sessionId = sessionId,
                        stageId = stageId,
                    ),
                )
            }

            // Record the tool execution outcome as events (invariant #5: every side effect is
            // captured; silent execution is not allowed). For file-mutating tools, materialise a
            // real file_written artifact from the actual on-disk result — so a stage that did NOT
            // write cannot have its file_written slot rubber-stamped as validated (F-015), and the
            // artifact carries the diff as evidence of the change.
            recordToolExecution(
                sessionId, stageId, toolCall, invocationId, tier, result,
                tool as? FileAffectingTool, request, fileWrittenSlots,
            )

            val sourceId = toolCall.id ?: invocationId.value
            val assistantEntry = ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L2,
                sourceType = "assistantToolCall",
                sourceId = sourceId,
                content = Json.encodeToString(ToolCallRequest.serializer(), toolCall),
                tokenEstimate = estimateTokens(toolCall.function.arguments),
                role = EntryRole.ASSISTANT,
            )
            val resultContent = when (result) {
                is ToolResult.Success ->
                    tool?.outputCompressor?.compress(result.output, ToolOutputContext(result.exitCode))
                        ?: result.output
                is ToolResult.Failure -> {
                    if (!result.recoverable) "FATAL: ${result.reason}"
                    else "ERROR: ${result.reason}"
                }
            }
            val resultEntry = ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L2,
                sourceType = "toolResult",
                sourceId = sourceId,
                content = resultContent,
                tokenEstimate = estimateTokens(resultContent),
                role = EntryRole.TOOL,
            )
            val steeringEntry = approvalNote?.takeIf { it.isNotBlank() }?.let {
                ContextEntry(
                    id = ContextEntryId(UUID.randomUUID().toString()),
                    layer = ContextLayer.L2,
                    sourceType = "steeringNote",
                    sourceId = sourceId,
                    content = "User steering on the approved '${toolCall.function.name}' call: $it",
                    tokenEstimate = estimateTokens(it),
                    role = EntryRole.USER,
                )
            }
            listOfNotNull(assistantEntry, resultEntry, steeringEntry)
        }
    }

    private suspend fun runPlane2Assessment(
        sessionId: SessionId,
        stageId: StageId,
        invocationId: ToolInvocationId,
        toolName: String,
        request: ToolRequest,
        tool: Tool?,
        effectives: RunEffectives,
        writeManifest: List<String>,
    ): RiskSummary? {
        val assessor = toolCallAssessor ?: return null
        val policy = effectives.policy ?: return null
        val assessment = assessor.assess(
            ToolCallAssessmentInput(
                request = request,
                capabilities = tool?.requiredCapabilities ?: emptySet(),
                workspace = policy,
                probe = worldProbe,
                paramRoles = tool?.paramRoles ?: emptyMap(),
                writeManifest = writeManifest,
            ),
        )
        emit(
            sessionId,
            ToolCallAssessedEvent(
                invocationId = invocationId,
                sessionId = sessionId,
                stageId = stageId,
                toolName = toolName,
                issues = assessment.toAssessedIssues(),
                observations = assessment.observations,
                disposition = assessment.disposition,
                timestampMs = Clock.System.now().toEpochMilliseconds(),
            ),
        )
        return assessment.toRiskSummary()
    }

    private suspend fun buildSchemaEntries(
        responseFormat: ResponseFormat,
        stageId: StageId,
    ): List<ContextEntry> {
        if (responseFormat !is ResponseFormat.Json) return emptyList()
        val compactSchema = Json.encodeToString(JsonSchema.serializer(), responseFormat.schema)
        val instruction = "Respond with a single JSON object matching this schema. " +
                "Do not include markdown, code fences, or commentary outside the JSON. " +
                "Schema: $compactSchema"
        return listOf(
            ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L0,
                content = instruction,
                sourceType = "schemaInstruction",
                sourceId = stageId.value,
                tokenEstimate = estimateTokens(instruction),
                role = EntryRole.SYSTEM,
            ),
        )
    }

    private suspend fun buildSteeringNoteEntries(sessionId: SessionId): List<ContextEntry> {
        val events = eventStore.read(sessionId)
        var locked = false
        return events.mapNotNull { event ->
            when (val p = event.payload) {
                is ExecutionPlanLockedEvent -> { locked = true; null }
                is SteeringNoteAddedEvent -> ContextEntry(
                    id = ContextEntryId(UUID.randomUUID().toString()),
                    layer = if (locked) ContextLayer.L0 else ContextLayer.L2,
                    content = if (locked) "PRIORITY DIRECTIVE — overrides the locked plan: ${p.content}"
                              else p.content,
                    sourceType = "steeringNote",
                    sourceId = p.stageId?.value ?: sessionId.value,
                    tokenEstimate = estimateTokens(p.content),
                    role = EntryRole.SYSTEM,
                )
                is ApprovalDecisionResolvedEvent -> {
                    val steering = p.userSteering
                    when {
                        steering != null -> ContextEntry(
                            id = ContextEntryId(UUID.randomUUID().toString()),
                            layer = ContextLayer.L2,
                            content = steering.text,
                            sourceType = "steeringNote",
                            sourceId = steering.sessionId.value,
                            tokenEstimate = estimateTokens(steering.text),
                            role = EntryRole.USER,
                        )
                        // A bare rejection (no note) still feeds back so the model does not
                        // re-propose the identical call on the retryable re-run (anti-loop).
                        p.outcome == ApprovalOutcome.REJECTED -> {
                            val feedback = "A previous tool call was rejected by the user. " +
                                "Do not repeat the same call; choose a different approach."
                            ContextEntry(
                                id = ContextEntryId(UUID.randomUUID().toString()),
                                layer = ContextLayer.L2,
                                content = feedback,
                                sourceType = "rejectionFeedback",
                                sourceId = sessionId.value,
                                tokenEstimate = estimateTokens(feedback),
                                role = EntryRole.SYSTEM,
                            )
                        }
                        else -> null
                    }
                }
                else -> null
            }
        }
    }

    /**
     * Injects the operator's answers to a stage's open questions as an L2 USER entry, so the stage
     * sees its own questions resolved on the clarification re-run. Correlates each answer's
     * questionId back to the prompt recorded on the [ClarificationRequestedEvent].
     */
    private suspend fun buildClarificationAnswerEntries(sessionId: SessionId): List<ContextEntry> {
        val events = eventStore.read(sessionId)
        val answered = events.mapNotNull { it.payload as? ClarificationAnsweredEvent }
        if (answered.isEmpty()) return emptyList()
        val prompts = events
            .mapNotNull { it.payload as? ClarificationRequestedEvent }
            .flatMap { it.questions }
            .associate { it.id to it.prompt }
        val rendered = answered.flatMap { it.answers }.joinToString("\n") { a ->
            "Q: ${prompts[a.questionId] ?: a.questionId}\nA: ${a.value}"
        }
        val content = "The operator answered your open questions. Treat these answers as authoritative " +
            "and do not ask them again:\n$rendered"
        return listOf(
            ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L2,
                content = content,
                sourceType = "clarificationAnswer",
                sourceId = sessionId.value,
                tokenEstimate = estimateTokens(content),
                role = EntryRole.USER,
            ),
        )
    }

    /**
     * If [stageId] just produced an LLM artifact carrying a non-empty `questions` array, park the
     * run: record the questions (invariant #9 — observed at parse time), await the operator's
     * answers, record them, and return true so the caller re-runs the stage with the answers in
     * context. Bounded by [MAX_CLARIFICATION_ROUNDS] so a stage that keeps re-asking eventually
     * proceeds. Returns false when there are no questions or the round budget is spent.
     */
    internal suspend fun requestClarificationIfNeeded(
        sessionId: SessionId,
        stageId: StageId,
        graph: WorkflowGraph,
    ): Boolean {
        val slot = graph.stages[stageId]?.produces?.firstOrNull { it.kind.llmEmitted } ?: return false
        val content = artifactContentCache["${sessionId.value}:${slot.name.value}"] ?: return false
        val questions = ClarificationQuestions.parse(content)
        if (questions.isEmpty()) return false

        val priorRounds = eventStore.read(sessionId)
            .count { (it.payload as? ClarificationRequestedEvent)?.stageId == stageId }
        if (priorRounds >= MAX_CLARIFICATION_ROUNDS) return false

        val requestId = ClarificationRequestId(UUID.randomUUID().toString())
        val deferred = CompletableDeferred<List<ClarificationAnswer>>()
        pendingClarifications[requestId] = deferred
        emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "CLARIFICATION_PENDING"))
        emit(sessionId, ClarificationRequestedEvent(requestId, sessionId, stageId, questions))
        return try {
            val answers = deferred.await()
            emit(sessionId, ClarificationAnsweredEvent(requestId, sessionId, stageId, answers))
            emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
            true
        } finally {
            pendingClarifications.remove(requestId)
        }
    }

    /**
     * Renders a top-K slice of the recorded repo map as a droppable L3 context entry. Reads
     * the latest [RepoMapComputedEvent] from the log (replay-safe — never re-scans the FS).
     * Not pinned: under budget pressure this yields before the working set (paths + symbols
     * only, so the model can `file_read` for detail).
     */
    private suspend fun buildRepoMapEntries(sessionId: SessionId): List<ContextEntry> {
        val map = eventStore.read(sessionId)
            .mapNotNull { it.payload as? RepoMapComputedEvent }
            .lastOrNull() ?: return emptyList()
        val top = map.entries.sortedByDescending { it.score }.take(REPO_MAP_INJECT_TOP_K)
        if (top.isEmpty()) return emptyList()
        val content = buildString {
            appendLine("## Repo map (top files by recency — read files for detail)")
            top.forEach { entry ->
                val symbols = if (entry.symbols.isEmpty()) "" else ": ${entry.symbols.joinToString(", ")}"
                appendLine("- ${entry.path}$symbols")
            }
        }.trimEnd()
        return listOf(
            ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L3,
                content = content,
                sourceType = "repoMap",
                sourceId = "repo-map",
                tokenEstimate = estimateTokens(content),
                role = EntryRole.SYSTEM,
            ),
        )
    }

    private suspend fun buildContextualRepoEntries(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ): List<ContextEntry> {
        val recorded = eventStore.read(sessionId)
            .mapNotNull { it.payload as? RepoKnowledgeRetrievedEvent }
            .lastOrNull { it.stageId == stageId }
        if (recorded != null) return listOf(buildRelevantFilesEntry(recorded.hits))

        val retriever = repoKnowledgeRetriever ?: return buildRepoMapEntries(sessionId)
        val query = (stageConfig.metadata["promptInline"] ?: stageConfig.metadata["prompt"] ?: stageId.value)
            .trim().ifBlank { stageId.value }
        val hits = runCatching { retriever.retrieve(sessionId, query, REPO_MAP_INJECT_TOP_K) }
            .getOrElse { e ->
                if (e is CancellationException) throw e
                log.warn("repo-knowledge retrieval failed for stage {}: {}", stageId.value, e.message)
                emptyList()
            }
        if (hits.isEmpty()) return buildRepoMapEntries(sessionId)
        emit(sessionId, RepoKnowledgeRetrievedEvent(sessionId, stageId, query, hits))
        return listOf(buildRelevantFilesEntry(hits))
    }

    private suspend fun buildNeedsArtifactEntries(
        sessionId: SessionId,
        stageConfig: StageConfig,
        criticIds: Set<ArtifactId> = emptySet(),
        criticFromStage: String? = null,
    ): List<ContextEntry> {
        return stageConfig.needs.mapNotNull { needed ->
            val cached = artifactContentCache["${sessionId.value}:${needed.value}"]
                ?.takeIf { it.isNotBlank() } ?: return@mapNotNull null
            val content = if (needed in criticIds) {
                "## Critic feedback from stage '${criticFromStage ?: "reviewer"}': ${needed.value}\n" +
                    "This is reviewer feedback on your previous attempt. " +
                    "Address every finding before completing the stage.\n$cached"
            } else {
                "## Input artifact: ${needed.value}\n$cached"
            }
            ContextEntry(
                id = ContextEntryId(UUID.randomUUID().toString()),
                layer = ContextLayer.L1,
                content = content,
                sourceType = if (needed in criticIds) "criticFeedback" else "neededArtifact",
                sourceId = needed.value,
                tokenEstimate = estimateTokens(content),
                role = EntryRole.USER,
            )
        }
    }

    private suspend fun emitToolArtifacts(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ) {
        stageConfig.produces
            .filter { !it.kind.llmEmitted }
            .forEach { slot ->
                // A file_written slot must be backed by real content materialised from an actual
                // successful write (see recordToolExecution). If none happened, do NOT fabricate a
                // validated artifact (F-015) — leave the slot unproduced so verifyProduces fails the
                // stage and it retries, rather than reporting a green run that changed nothing.
                if (slot.kind.id == "file_written" &&
                    artifactContentCache["${sessionId.value}:${slot.name.value}"] == null
                ) {
                    return@forEach
                }
                emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1))
                emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId))
                emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId))
            }
    }

    private suspend fun recordToolExecution(
        sessionId: SessionId,
        stageId: StageId,
        toolCall: ToolCallRequest,
        invocationId: ToolInvocationId,
        tier: Tier,
        result: ToolResult,
        fileTool: FileAffectingTool?,
        request: ToolRequest,
        fileWrittenSlots: List<TypedArtifactSlot>,
    ) {
        when (result) {
            is ToolResult.Failure -> emit(
                sessionId,
                ToolExecutionFailedEvent(invocationId, sessionId, toolCall.function.name, result.reason),
            )

            is ToolResult.Success -> {
                val diff = result.metadata["diff"]?.takeIf { it.isNotBlank() }
                val affected = fileTool?.affectedPaths(request).orEmpty()
                emit(
                    sessionId,
                    ToolExecutionCompletedEvent(
                        invocationId = invocationId,
                        sessionId = sessionId,
                        toolName = toolCall.function.name,
                        receipt = ToolReceipt(
                            invocationId = invocationId,
                            toolName = toolCall.function.name,
                            exitCode = result.exitCode,
                            outputSummary = result.output.take(OUTPUT_SUMMARY_LIMIT),
                            affectedEntities = affected.map { it.toString() },
                            durationMs = 0,
                            tier = tier,
                            timestamp = Clock.System.now(),
                            diff = diff,
                        ),
                    ),
                )
                if (affected.isNotEmpty()) {
                    materializeFileWritten(sessionId, stageId, request, affected, fileWrittenSlots, diff)
                    affected.forEach { p ->
                        emit(
                            sessionId,
                            FileWrittenEvent(
                                invocationId = invocationId,
                                sessionId = sessionId,
                                path = p.toString(),
                                preExisted = true,
                                timestampMs = Clock.System.now().toEpochMilliseconds(),
                            ),
                        )
                    }
                }
            }
        }
    }

    /**
     * Materialise a [FileWrittenArtifact] from the actual on-disk file after a successful mutation,
     * store it as the stage's file_written slot content, and record [ArtifactContentStoredEvent].
     * This ties the file_written artifact to a real write so validation can no longer be faked.
     */
    private suspend fun materializeFileWritten(
        sessionId: SessionId,
        stageId: StageId,
        request: ToolRequest,
        affected: Set<java.nio.file.Path>,
        fileWrittenSlots: List<TypedArtifactSlot>,
        diff: String?,
    ) {
        if (fileWrittenSlots.isEmpty()) return
        val relPath = request.parameters["path"] as? String ?: return
        val onDisk = affected.firstOrNull() ?: Paths.get(relPath)
        val bytes = runCatching { Files.readAllBytes(onDisk) }.getOrNull() ?: return
        val contentHash = artifactStore.put(bytes)
        val artifact = FileWrittenArtifact(
            path = relPath,
            contentHash = contentHash.value,
            size = bytes.size.toLong(),
            mode = "0644",
            diff = diff,
        )
        val json = Json.encodeToString(FileWrittenArtifact.serializer(), artifact)
        val storedHash = artifactStore.put(json.toByteArray())
        fileWrittenSlots.forEach { slot ->
            artifactContentCache["${sessionId.value}:${slot.name.value}"] = json
            emit(
                sessionId,
                ArtifactContentStoredEvent(
                    artifactId = slot.name,
                    contentHash = storedHash,
                    sessionId = sessionId,
                    stageId = stageId,
                ),
            )
        }
    }

    private suspend fun emitLlmArtifacts(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ) {
        stageConfig.produces
            .filter { it.kind.llmEmitted }
            .forEach { slot ->
                // Same rule as the file_written gate (F-015): no stored content means the
                // model never produced this artifact — do not fabricate a validated one.
                val cached = artifactContentCache["${sessionId.value}:${slot.name.value}"]
                if (cached.isNullOrBlank()) {
                    log.warn(
                        "[Orchestrator] stage={} slot={}: no content stored — skipping artifact emission",
                        stageId.value,
                        slot.name.value,
                    )
                    return@forEach
                }
                emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1))
                emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId))
                emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId))
            }
    }

    private fun verifyProduces(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ): StageExecutionResult {
        if (stageConfig.produces.isEmpty()) {
            if (stageConfig.allowedTools.isEmpty()) return StageExecutionResult.Success(emptyList())
            val ranTool = eventStore.read(sessionId).any {
                (it.payload as? ToolInvocationRequestedEvent)?.stageId == stageId
            }
            if (ranTool) return StageExecutionResult.Success(emptyList())
            log.error(
                "[Orchestrator] stage declared no artifacts and ran no tools " +
                        "session=${sessionId.value} stage=${stageId.value}",
            )
            return StageExecutionResult.Failure(
                "stage ${stageId.value} declared no artifacts and ran no tools",
                retryable = true,
            )
        }
        val producedIds = stageConfig.produces.map { it.name }.toSet()
        val present = eventStore.read(sessionId)
            .mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId }
            .toSet()
        val missing = producedIds - present
        if (missing.isEmpty()) return StageExecutionResult.Success(emptyList())
        val missingIds = missing.joinToString(", ") { it.value }
        log.error(
            "[Orchestrator] stage missing produced artifacts " +
                    "session=${sessionId.value} stage=${stageId.value} missing=$missingIds",
        )
        return StageExecutionResult.Failure(
            "stage ${stageId.value} did not produce declared artifacts: $missingIds",
            retryable = true,
        )
    }

    /**
     * Stage-level plan checkpointing (BACKLOG §C-A2): make plan adherence observable and replayable
     * by emitting a first-class checkpoint event per stage, reconciling produced-vs-expected against
     * the confirmed (locked) plan. Only runs on plan-driven (phase-2) sessions — i.e. a log that
     * holds an [ExecutionPlanLockedEvent]; non-plan workflows emit nothing. The halt itself is owned
     * by [verifyProduces]; this method only EMITS — a [StageCheckpointFailedEvent] accompanies that
     * halt and surfaces *why* the plan diverged.
     */
    private suspend fun emitStageCheckpoint(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ) {
        if (eventStore.read(sessionId).none { it.payload is ExecutionPlanLockedEvent }) return
        val expected = stageConfig.produces.map { it.name.value }.toSet()
        if (expected.isEmpty()) return
        val produced = eventStore.read(sessionId)
            .mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId?.value }
            .toSet()
        val cp = StageCheckpointReconciler.reconcile(expected, produced)
        if (cp.passed) {
            emit(sessionId, StageCheckpointPassedEvent(sessionId, stageId, expected, produced))
        } else {
            emit(sessionId, StageCheckpointFailedEvent(sessionId, stageId, expected, produced, cp.missing))
        }
    }

    /**
     * Entity grounding (role-reliability §3): for a stage that opted in (`ground_references`),
     * check every workspace-relative file path its brief named against the filesystem. Existence
     * is a nondeterministic observation, so the result is recorded as a [BriefGroundingCheckedEvent]
     * (invariant #9) — replay reads it back, never re-probes. A hallucinated path fails the stage
     * (retryable), feeding the missing paths back so the model re-grounds on the next attempt.
     */
    private suspend fun groundBriefReferences(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
        effectives: RunEffectives,
    ): StageExecutionResult {
        if (stageConfig.metadata["groundReferences"] != "true") return StageExecutionResult.Success(emptyList())
        val workspaceRoot = effectives.policy?.workspaceRoot ?: return StageExecutionResult.Success(emptyList())
        val artifact = stageConfig.produces
            .filter { it.kind.llmEmitted }
            .firstNotNullOfOrNull { artifactContentCache["${sessionId.value}:${it.name.value}"] }
            ?: return StageExecutionResult.Success(emptyList())

        val references = BriefReferenceExtractor.fileReferences(artifact)
        if (references.isEmpty()) return StageExecutionResult.Success(emptyList())

        val results = references.map { ref ->
            GroundingReference(ref, worldProbe.exists(workspaceRoot.resolve(ref)))
        }
        emit(sessionId, BriefGroundingCheckedEvent(sessionId, stageId, stageId.value, results))

        val missing = results.filterNot { it.grounded }.map { it.reference }
        return if (missing.isEmpty()) {
            StageExecutionResult.Success(emptyList())
        } else {
            log.warn(
                "[Orchestrator] brief references nonexistent paths session={} stage={} missing={}",
                sessionId.value, stageId.value, missing.joinToString(", "),
            )
            StageExecutionResult.Failure(
                "stage ${stageId.value} references paths that do not exist in the workspace: " +
                    "${missing.joinToString(", ")}. Reference only files you have read.",
                retryable = true,
            )
        }
    }

    /**
     * Brief echo-back gate (BACKLOG §C-A1): for a stage that opted in (`briefEcho`), the planner must
     * restate the brief as a structured `brief_echo` block before any plan is generated. The echo is
     * diffed against the original brief; on divergence a [BriefEchoMismatchEvent] is emitted and the
     * stage fails (retryable), so the plan is never accepted — the cheapest failure point in the
     * pipeline. Mismatch is conservative ([BriefEchoComparator]): a dropped acceptance criterion or an
     * invented file path halts; merely dropping a file/symbol does not.
     */
    private suspend fun checkBriefEcho(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ): StageExecutionResult {
        if (stageConfig.metadata["briefEcho"] != "true") return StageExecutionResult.Success(emptyList())
        val sourceSlot = stageConfig.metadata["briefEchoSource"] ?: "brief"
        val original = artifactContentCache["${sessionId.value}:$sourceSlot"]
            ?: return StageExecutionResult.Success(emptyList())

        val produced = stageConfig.produces
            .filter { it.kind.llmEmitted }
            .firstNotNullOfOrNull { artifactContentCache["${sessionId.value}:${it.name.value}"] }
            ?: return StageExecutionResult.Success(emptyList())

        val echo = BriefEchoExtractor.fromEcho(produced)
            ?: return StageExecutionResult.Failure(
                "stage ${stageId.value} must restate the brief as a brief_echo block before planning",
                retryable = true,
            )

        val diff = BriefEchoComparator.compare(BriefEchoExtractor.fromOriginalBrief(original), echo)
        if (!diff.hasMismatch) return StageExecutionResult.Success(emptyList())

        emit(
            sessionId,
            BriefEchoMismatchEvent(
                sessionId = sessionId,
                stageId = stageId,
                droppedFiles = diff.droppedFiles,
                droppedSymbols = diff.droppedSymbols,
                droppedAcceptanceCriteria = diff.droppedAcceptanceCriteria,
                inventedFiles = diff.inventedFiles,
            ),
        )
        log.warn(
            "[Orchestrator] brief echo diverged session={} stage={} droppedCriteria={} invented={}",
            sessionId.value, stageId.value,
            diff.droppedAcceptanceCriteria.joinToString(", "),
            diff.inventedFiles.joinToString(", "),
        )
        return StageExecutionResult.Failure(
            buildString {
                append("stage ${stageId.value} brief_echo diverged from the brief.")
                if (diff.droppedAcceptanceCriteria.isNotEmpty()) {
                    append(" Restate these acceptance criteria: ${diff.droppedAcceptanceCriteria.joinToString(", ")}.")
                }
                if (diff.inventedFiles.isNotEmpty()) {
                    append(" These files are not in the brief — do not invent paths: " +
                        "${diff.inventedFiles.joinToString(", ")}.")
                }
            },
            retryable = true,
        )
    }

    private suspend fun emitProcessResultEvents(
        sessionId: SessionId,
        stageId: StageId,
        stageConfig: StageConfig,
    ) {
        // Emit artifact lifecycle events for process_result slots on failure branches.
        // Content was already stored in CAS + cache by dispatchToolCalls for every outcome.
        stageConfig.produces.filter { it.kind.id == "process_result" }.forEach { slot ->
            emit(sessionId, ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1))
            emit(sessionId, ArtifactValidatingEvent(slot.name, sessionId, stageId))
            emit(sessionId, ArtifactValidatedEvent(slot.name, sessionId, stageId))
        }
    }

    internal open suspend fun mapValidationOutcome(
        sessionId: SessionId,
        stageId: StageId,
        context: ValidationContext,
    ): StageExecutionResult = when (val outcome = validationPipeline.validate(context)) {
        is ValidationOutcome.Passed -> StageExecutionResult.Success(emptyList())
        is ValidationOutcome.Rejected ->
            StageExecutionResult.Failure("validation failed", retryable = outcome.retryable)

        is ValidationOutcome.NeedsApproval -> handleApproval(sessionId, stageId, outcome)
    }

    @Suppress("LongParameterList")
    internal open suspend fun runInference(
        sessionId: SessionId,
        stageId: StageId,
        contextPack: ContextPack,
        stageConfig: StageConfig,
        timeoutMs: Long,
        responseFormat: ResponseFormat = ResponseFormat.Text,
        effectives: RunEffectives = RunEffectives(toolRegistry, toolExecutor, workspacePolicy),
    ): InferenceResult {
        val provider = inferenceRouter.route(stageId, stageConfig.requiredCapabilities, stageConfig.modelId)
        log.debug(
            "[Orchestrator] inference session={} stage={} provider={} timeoutMs={}",
            sessionId.value, stageId.value, provider.id.value, timeoutMs,
        )
        val requestId = InferenceRequestId(UUID.randomUUID().toString())
        val request = InferenceRequest(
            requestId = requestId,
            sessionId = sessionId,
            stageId = stageId,
            contextPack = contextPack,
            generationConfig = stageConfig.generationConfig,
            responseFormat = responseFormat,
            tools = stageConfig.allowedTools
                .mapNotNull { effectives.registry?.resolve(it) }
                .map { tool ->
                    ToolDefinition(
                        function = ToolFunction(
                            name = tool.name,
                            description = tool.description,
                            parameters = tool.parametersSchema,
                        ),
                    )
                } + ToolDefinition(
                    function = ToolFunction(
                        name = STAGE_COMPLETE_TOOL,
                        description = "Call this tool when the stage's goal is fully met and no further tool calls are needed. " +
                                "The orchestrator will proceed to the next stage.",
                        parameters = JsonObject(emptyMap()),
                    ),
                ),
        )

        val renderedMessages = PromptRenderer.render(contextPack)
        val promptArtifactId = artifactStore.put(Json.encodeToString(renderedMessages).toByteArray())
        emit(sessionId, InferenceStartedEvent(requestId, sessionId, stageId, provider.id, promptArtifactId))

        if (isCancelled(sessionId)) return InferenceResult.Cancelled

        return try {
            val rawResponse = withTimeout(timeoutMs) { provider.infer(request) }
            // Models frequently wrap a JSON artifact in a ```json … ``` markdown fence despite
            // being told not to. The fence makes the artifact unparseable → validation fails.
            // Strip an outer fence wrapper so the stored artifact, its CAS hash, the content
            // cache, and validation all see the same clean payload.
            val response = rawResponse.copy(text = stripCodeFence(rawResponse.text))
            log.debug(
                "[Orchestrator] inference done session={} stage={} tokens={} latencyMs={}",
                sessionId.value, stageId.value, response.tokensUsed, response.latencyMs,
            )

            if (isCancelled(sessionId)) return InferenceResult.Cancelled

            val responseArtifactId = artifactStore.put(response.text.toByteArray())
            emit(
                sessionId,
                InferenceCompletedEvent(
                    requestId = requestId,
                    sessionId = sessionId,
                    stageId = stageId,
                    providerId = provider.id,
                    tokensUsed = response.tokensUsed,
                    latencyMs = response.latencyMs,
                    responseArtifactId = responseArtifactId,
                ),
            )

            InferenceResult.Success(response, responseArtifactId)
        } catch (e: TimeoutCancellationException) {
            emit(
                sessionId,
                InferenceTimeoutEvent(
                    requestId = requestId,
                    sessionId = sessionId,
                    stageId = stageId,
                    providerId = provider.id,
                    timeoutMs = timeoutMs,
                ),
            )
            InferenceResult.Failed("TIMEOUT: ${e.message}")
        } catch (e: CancellationException) {
            throw e // never swallow
        } catch (e: Exception) {
            emit(
                sessionId,
                InferenceFailedEvent(
                    requestId = requestId,
                    sessionId = sessionId,
                    stageId = stageId,
                    providerId = provider.id,
                    reason = e.message ?: "unknown",
                ),
            )
            InferenceResult.Failed(e.message ?: "inference failed")
        }
    }

    // A provider 4xx client error (e.g. llama.cpp 400 "invalid_request") is a deterministic
    // request-construction failure — retrying re-sends the identical bad request and exhausts the
    // budget for nothing (F-002). Treat 4xx as terminal, except 408 (timeout) and 429 (rate-limit)
    // which are transient. Everything else (network/5xx/unknown) stays retryable.
    private fun isRetryableFailure(reason: String): Boolean {
        val status = Regex("""returned\s+(\d{3})""").find(reason)?.groupValues?.get(1)?.toIntOrNull()
            ?: return true
        return status !in 400..499 || status == HTTP_TIMEOUT || status == HTTP_TOO_MANY_REQUESTS
    }

    // --- transition helpers ---

    internal fun resolveTransition(
        graph: WorkflowGraph,
        sessionId: SessionId,
        currentStageId: StageId,
        artifacts: Map<ArtifactId, ArtifactState> = emptyMap(),
        artifactContent: Map<ArtifactId, String> = emptyMap(),
    ): TransitionDecision {
        val ctx = EvaluationContext(
            sessionId = sessionId,
            currentStage = currentStageId,
            artifacts = artifacts,
            artifactContent = artifactContent,
        )
        return transitionResolver.resolve(graph, ctx)
    }

    internal fun isTerminal(graph: WorkflowGraph, stageId: StageId): Boolean =
        graph.transitions.none { it.from == stageId }

    internal suspend fun advanceStage(
        sessionId: SessionId,
        fromStageId: StageId,
        decision: TransitionDecision.Move,
    ): StageId {
        emit(
            sessionId,
            TransitionExecutedEvent(sessionId, fromStageId, decision.to, decision.transitionId, decision.redirectId),
        )
        return decision.to
    }

    // --- terminal state helpers ---

    internal suspend fun emitWorkflowStarted(sessionId: SessionId, graph: WorkflowGraph, config: OrchestrationConfig) {
        emit(
            sessionId,
            WorkflowStartedEvent(
                sessionId = sessionId,
                workflowId = graph.id,
                startStageId = graph.start,
                retryPolicy = config.retryPolicy,
            ),
        )
    }

    internal suspend fun completeWorkflow(
        sessionId: SessionId,
        terminalStageId: StageId,
        stageCount: Int,
    ): WorkflowResult.Completed {
        log.debug(
            "[Orchestrator] COMPLETED session={} terminalStage={} stages={}",
            sessionId.value, terminalStageId.value, stageCount,
        )
        emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount))
        cancellations.remove(sessionId)
        return WorkflowResult.Completed(sessionId, terminalStageId)
    }

    internal suspend fun failWorkflow(
        sessionId: SessionId,
        stageId: StageId,
        reason: String,
        retryExhausted: Boolean,
    ): WorkflowResult.Failed {
        log.warn(
            "[Orchestrator] FAILED session={} stage={} reason={} retryExhausted={}",
            sessionId.value, stageId.value, reason, retryExhausted,
        )
        emit(sessionId, WorkflowFailedEvent(sessionId, stageId, reason, retryExhausted))
        cancellations.remove(sessionId)
        return WorkflowResult.Failed(sessionId, reason, retryExhausted)
    }

    internal suspend fun handleCancellation(
        sessionId: SessionId,
        stageId: StageId,
    ): WorkflowResult.Cancelled {
        log.warn("[Orchestrator] CANCELLED session={} stage={}", sessionId.value, stageId.value)
        emit(sessionId, WorkflowFailedEvent(sessionId, stageId, "CANCELLED", retryExhausted = false))
        cancellations.remove(sessionId)
        return WorkflowResult.Cancelled(sessionId)
    }

    // --- cancellation ---

    internal fun isCancelled(sessionId: SessionId): Boolean =
        cancellations[sessionId]?.get() == true

    // --- event emission ---

    // Surface context-budget overflow as an event on the workflow path. The builder already
    // enforces the budget (drops oldest/compressible entries, pinning steering + event history);
    // this records that it happened so replay and operators can see it, mirroring the router path.
    private suspend fun emitContextTruncationIfNeeded(
        sessionId: SessionId,
        stageId: StageId,
        contextPack: ContextPack,
    ) {
        val meta = contextPack.compressionMetadata
        if (meta.entriesDropped <= 0) return
        emit(
            sessionId,
            ContextTruncatedEvent(
                sessionId = sessionId,
                turnId = stageId.value,
                entriesDropped = meta.entriesDropped,
                truncatedLayers = meta.truncatedLayers.map { it.name },
                timestampMs = Clock.System.now().toEpochMilliseconds(),
            ),
        )
    }

    internal suspend fun emit(sessionId: SessionId, payload: EventPayload) {
        log.debug("[session {}] emitting event, payload: {}", sessionId.value.take(7), payload)
        eventStore.append(
            NewEvent(
                metadata = EventMetadata(
                    eventId = EventId(UUID.randomUUID().toString()),
                    sessionId = sessionId,
                    timestamp = Clock.System.now(),
                    schemaVersion = 1,
                    causationId = null,
                    correlationId = null,
                ),
                payload = payload,
            ),
        )
    }

    // --- token estimation ---

    protected open suspend fun estimateTokens(content: String): Int {
        val t = tokenizer
        if (t != null) {
            // A zero count for non-blank content means the tokenizer endpoint is lying
            // (e.g. a malformed request silently answered with an empty token list) —
            // trust the heuristic instead, or every budget check goes blind.
            return runCatching { t.countTokens(content) }
                .getOrElse { fallbackTokenEstimate(content) }
                .takeIf { it > 0 || content.isBlank() }
                ?: fallbackTokenEstimate(content)
        }
        return fallbackTokenEstimate(content)
    }

    private fun fallbackTokenEstimate(content: String): Int {
        return (content.length / 4).coerceAtLeast(1)
    }

    // --- stage-entry approval gate ---

    /**
     * Emits [OrchestrationPausedEvent] + [ApprovalRequestedEvent] for a stage flagged
     * `requiresApproval`, awaits the user decision, and records it.
     * Returns `true` if the run should proceed into the stage, `false` if the approval was rejected.
     */
    internal suspend fun requestStageApproval(
        sessionId: SessionId,
        stageId: StageId,
        preview: String,
    ): Boolean {
        val requestId = ApprovalRequestId(UUID.randomUUID().toString())
        val validationReportId = ValidationReportId(UUID.randomUUID().toString())
        val deferred = CompletableDeferred<ApprovalDecision>()
        pendingApprovals[requestId] = deferred

        emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING"))
        emit(
            sessionId,
            ApprovalRequestedEvent(
                requestId = requestId,
                tier = Tier.T2,
                validationReportId = validationReportId,
                riskSummaryId = null,
                riskSummary = null,
                sessionId = sessionId,
                stageId = stageId,
                projectId = null,
                preview = preview,
            ),
        )

        return try {
            val decision = deferred.await()
            emitDecisionResolved(
                sessionId,
                DomainApprovalRequest(
                    id = requestId,
                    tier = Tier.T2,
                    validationReportId = validationReportId,
                    riskSummaryId = null,
                    riskSummary = null,
                    timestamp = Clock.System.now(),
                ),
                decision,
            )
            if (decision.isApproved) {
                emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
                true
            } else {
                emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
                false
            }
        } finally {
            pendingApprovals.remove(requestId)
        }
    }

    /** Public seam for out-of-band gates (e.g. freestyle plan review) that reuse the stage-approval flow. */
    suspend fun requestPlanApproval(sessionId: SessionId, preview: String): Boolean =
        requestStageApproval(sessionId, StageId("plan_review"), preview)

    // --- private functions ---

    private suspend fun handleApproval(
        sessionId: SessionId,
        stageId: StageId,
        outcome: ValidationOutcome.NeedsApproval,
    ): StageExecutionResult {
        log.debug("[Orchestrator] approval required session={} stage={}", sessionId.value, stageId.value)
        val domainRequest = buildApprovalRequest(sessionId, stageId, outcome)
        val deferred = CompletableDeferred<ApprovalDecision>()
        pendingApprovals[domainRequest.id] = deferred

        emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING"))
        emit(
            sessionId,
            ApprovalRequestedEvent(
                requestId = domainRequest.id,
                tier = domainRequest.tier,
                validationReportId = domainRequest.validationReportId,
                riskSummaryId = domainRequest.riskSummaryId,
                riskSummary = domainRequest.riskSummary,
                sessionId = sessionId,
                stageId = stageId,
                projectId = null,
            ),
        )

        return try {
            log.debug(
                "[Orchestrator] awaiting approval session={} requestId={}",
                sessionId.value, domainRequest.id.value,
            )
            val decision = deferred.await()
            log.debug(
                "[Orchestrator] approval resolved session={} approved={}",
                sessionId.value, decision.isApproved,
            )
            emitDecisionResolved(sessionId, domainRequest, decision)
            if (decision.isApproved) {
                emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
                StageExecutionResult.Success(emptyList())
            } else {
                emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
                StageExecutionResult.Failure(decision.reason ?: "approval rejected", retryable = false)
            }
        } finally {
            pendingApprovals.remove(domainRequest.id)
        }
    }

    private suspend fun buildApprovalRequest(
        sessionId: SessionId,
        stageId: StageId,
        outcome: ValidationOutcome.NeedsApproval,
    ): DomainApprovalRequest {
        val state = orchestrationRepository.getState(sessionId)
        val inferenceState = inferenceRepository.getInferenceState(sessionId)
        val riskSummary = riskAssessor.assess(
            RiskContext(
                validationReport = outcome.request.validationReport,
                orchestrationState = state,
                inferenceState = inferenceState,
            ),
        )
        val riskSummaryId = RiskSummaryId(UUID.randomUUID().toString())
        // RiskAssessedEvent records the indexed level+action summary; ApprovalRequestedEvent carries the full
        // authoritative RiskSummary (signals + rationale). Both derive from the same assessment value so
        // they cannot diverge.
        emit(
            sessionId,
            RiskAssessedEvent(sessionId, stageId, riskSummaryId, riskSummary.level, riskSummary.recommendedAction),
        )
        return DomainApprovalRequest(
            id = ApprovalRequestId(UUID.randomUUID().toString()),
            tier = riskSummary.level.toApprovalTier(),
            validationReportId = ValidationReportId(UUID.randomUUID().toString()),
            riskSummaryId = riskSummaryId,
            riskSummary = riskSummary,
            timestamp = Clock.System.now(),
        )
    }

    private suspend fun emitDecisionResolved(
        sessionId: SessionId,
        domainRequest: DomainApprovalRequest,
        decision: ApprovalDecision,
    ) {
        emit(
            sessionId,
            ApprovalDecisionResolvedEvent(
                decisionId = ApprovalDecisionId(UUID.randomUUID().toString()),
                requestId = domainRequest.id,
                outcome = decision.outcome ?: ApprovalOutcome.REJECTED,
                status = ApprovalStatus.COMPLETED,
                tier = domainRequest.tier,
                resolutionTimestamp = Clock.System.now(),
                reason = decision.reason,
                userSteering = decision.userSteering,
            ),
        )
    }
}

/**
 * Compute a unified-diff preview for file-affecting tool calls.
 * For [file_write] with operation "write": reads the existing file (if any) and diffs it
 * against the proposed content from the LLM arguments.
 * For all other tools: returns null (caller falls back to raw JSON args).
 *
 * This function performs blocking I/O (file reads) and must be called from a suspend context
 * that will dispatch it on [Dispatchers.IO].
 */
private suspend fun computeToolPreview(toolName: String, parameters: Map<String, Any>): String? {
    if (toolName == "shell") return shellCommandPreview(parameters)
    if (toolName != "file_write") return null
    val path = parameters["path"] as? String ?: return null
    val operation = parameters["operation"] as? String
    if (operation != "write") return null
    val proposedContent = parameters["content"] as? String ?: return null

    val existingContent = withContext(Dispatchers.IO) {
        runCatching {
            val filePath = java.nio.file.Paths.get(path)
            if (java.nio.file.Files.exists(filePath)) {
                java.nio.file.Files.readString(filePath)
            } else null
        }.getOrNull()
    }

    return buildDiffString(path, existingContent, proposedContent)
}

/**
 * Render a shell tool's argv as a full, shell-quoted command line for the approval
 * card (tui-requirements §5 / §3 "full untruncated command"). Returning the rendered
 * line — rather than the raw `arguments.take(200)` fallback — keeps the command both
 * untruncated and human-readable; the TUI's deterministic parser re-tokenizes it.
 */
private fun shellCommandPreview(parameters: Map<String, Any>): String? {
    val argv = when (val raw = parameters["argv"]) {
        is List<*> -> raw.filterIsInstance<String>()
        is String -> runCatching { Json.decodeFromString<List<String>>(raw) }.getOrElse { emptyList() }
        else -> emptyList()
    }
    if (argv.isEmpty()) return null
    return argv.joinToString(" ") { shellQuoteToken(it) }
}

/** Single-quote a token when it carries whitespace or shell metacharacters. */
private fun shellQuoteToken(token: String): String {
    if (token.isEmpty()) return "''"
    val needsQuote = token.any { it.isWhitespace() || it in "\"'\\|&;<>(){}\$`*?[]~#!" }
    return if (needsQuote) "'" + token.replace("'", "'\\''") + "'" else token
}

/**
 * Build a unified-diff-style string for a full file replacement.
 * When the file doesn't exist yet (new file), only `+` lines are shown.
 * Uses a simple full-file diff: all old lines as `-`, all new lines as `+`.
 */
private fun buildDiffString(path: String, existingContent: String?, proposedContent: String): String {
    val existingLines = existingContent?.lines() ?: emptyList()
    val proposedLines = proposedContent.lines()
    if (existingLines == proposedLines) return "  (no change)"

    return buildString {
        appendLine("--- a/$path")
        appendLine("+++ b/$path")
        appendLine("@@ -1,${existingLines.size.coerceAtLeast(1)} +1,${proposedLines.size.coerceAtLeast(1)} @@")
        existingLines.forEach { appendLine("-$it") }
        proposedLines.forEach { appendLine("+$it") }
    }.trimEnd()
}

internal sealed interface InferenceResult {
    data class Success(val response: InferenceResponse, val responseArtifactId: ArtifactId? = null) : InferenceResult
    data class Failed(val reason: String) : InferenceResult
    data object Cancelled : InferenceResult
}
