refactor(kernel): decompose SessionOrchestrator god-class into per-concern extensions

Splits the 3.7k-line SessionOrchestrator into the state-owning abstract class
(fields + open/abstract seams: run, cancel, runInference, mapValidationOutcome,
estimateTokens) plus behavior-preserving internal extension-fun files grouped by
concern: Artifacts, ToolExec, Workspace, Context, RepoContext, Gates, Gates2,
Workflow, Approval, Preview. Two public members consumed cross-module
(liveClarificationRequestIds, requestPlanApproval) stay on the class. Each file
kept <=10 top-level funs; pure relocation, no logic changes.

Clears LargeClass; adds no new TooManyFunctions. kernel test + detekt green;
server/cli compile clean.
This commit is contained in:
2026-07-13 11:58:44 +04:00
parent 7f820bafe6
commit 670e0c4828
12 changed files with 2962 additions and 2897 deletions
@@ -0,0 +1,195 @@
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.model.ApprovalDecision
import com.correx.core.approvals.model.DomainApprovalRequest
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
import com.correx.core.events.events.ApprovalRequestedEvent
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.risk.RiskSummary
import com.correx.core.events.types.ApprovalDecisionId
import com.correx.core.events.types.ApprovalRequestId
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.ValidationReportId
import com.correx.core.sessions.ApprovalMode
import com.correx.core.risk.RiskContext
import com.correx.core.risk.toApprovalTier
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.validation.pipeline.ValidationOutcome
import kotlinx.coroutines.CompletableDeferred
import kotlinx.datetime.Clock
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
@Suppress("CyclomaticComplexMethod")
// Maps the operator profile's free-text approval_mode onto the engine's [ApprovalMode].
// Unset/unknown falls back to PROMPT so an absent profile keeps the human-in-the-loop
// default; only an explicit auto/yolo opts a session into unattended approval.
internal fun SessionOrchestrator.approvalModeFor(profileMode: String?): ApprovalMode =
when (profileMode?.trim()?.lowercase()) {
"deny" -> ApprovalMode.DENY
"auto" -> ApprovalMode.AUTO
"yolo" -> ApprovalMode.YOLO
else -> ApprovalMode.PROMPT
}
internal suspend fun SessionOrchestrator.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)
}
}
// --- private functions ---
internal suspend fun SessionOrchestrator.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)
}
}
internal suspend fun SessionOrchestrator.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(),
)
}
internal suspend fun SessionOrchestrator.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,
),
)
}
@@ -0,0 +1,252 @@
package com.correx.core.kernel.orchestration
import com.correx.core.artifacts.kind.JsonSchema
import com.correx.core.artifacts.kind.FileWrittenArtifact
import com.correx.core.artifacts.kind.TypedArtifactSlot
import com.correx.core.context.model.ContextBucket
import com.correx.core.context.model.ContextEntry
import com.correx.core.context.model.ContextLayer
import com.correx.core.context.model.TokenBudget
import com.correx.core.context.model.EntryRole
import com.correx.core.events.events.ArtifactContentStoredEvent
import com.correx.core.events.events.FileWrittenEvent
import com.correx.core.events.events.ArtifactCreatedEvent
import com.correx.core.events.events.ArtifactRepairAttemptedEvent
import com.correx.core.events.events.ArtifactRepairFailedEvent
import com.correx.core.events.events.ArtifactRepairResolvedEvent
import com.correx.core.events.events.BriefGroundingCheckedEvent
import com.correx.core.events.events.ExecutionPlanLockedEvent
import com.correx.core.events.events.StageCheckpointFailedEvent
import com.correx.core.events.events.StageCheckpointPassedEvent
import com.correx.core.events.events.ToolInvocationRequestedEvent
import com.correx.core.toolintent.WorkspacePolicy
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.ContextEntryId
import com.correx.core.events.types.ContextPackId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.inference.ResponseFormat
import com.correx.core.inference.ToolDefinition
import com.correx.core.inference.ToolFunction
import com.correx.core.tools.contract.Tool
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.transitions.graph.StageConfig
import com.correx.core.validation.artifact.ArtifactExtractionPipeline
import com.correx.core.validation.artifact.ArtifactFailure
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.jsonObject
import java.nio.file.Files
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
// --- per-run effective registry + policy ---
internal fun SessionOrchestrator.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)
}
/**
* Handles an [ArtifactExtractionPipeline.ExtractionResult.Unresolved]: attempts one grammar-constrained
* LLM-repair rung (only for SCHEMA / MISSING_INFO — the classifier gates the expensive rung), then
* consults the policy [decide] to map a hard failure onto an orchestrator action. Records the §6 events.
*/
internal suspend fun SessionOrchestrator.repairArtifact(
sessionId: SessionId,
stageId: StageId,
slot: TypedArtifactSlot,
unresolved: ArtifactExtractionPipeline.ExtractionResult.Unresolved,
stageConfig: StageConfig,
effectives: RunEffectives,
timeoutMs: Long,
): ArtifactLadderOutcome {
val schema = slot.kind.deriveJsonSchema()
val eligible = unresolved.classification in setOf(ArtifactFailure.SCHEMA, ArtifactFailure.MISSING_INFO)
val bestCandidate = unresolved.bestCandidate
if (eligible && bestCandidate != null && !isCancelled(sessionId)) {
emitArtifactRepairAttempted(sessionId, stageId, slot, unresolved.classification, "LLM")
val repaired = runArtifactRepairInference(
sessionId, stageId, slot, bestCandidate, stageConfig, effectives, timeoutMs,
)
val reRun = repaired?.let { artifactExtractionPipeline.run(it, schema) }
if (reRun is ArtifactExtractionPipeline.ExtractionResult.Resolved) {
val json = reRun.canonicalJson.toString()
emitArtifactRepairResolved(sessionId, stageId, slot, json)
return ArtifactLadderOutcome.Text(json)
}
}
val decision = decide(unresolved.classification, attemptsSoFar = 0, maxAttempts = Int.MAX_VALUE, hasApprover = false)
emit(
sessionId,
ArtifactRepairFailedEvent(
artifactId = slot.name,
classification = unresolved.classification.name,
decision = decision::class.simpleName ?: "Unknown",
reason = unresolved.detail,
sessionId = sessionId,
stageId = stageId,
),
)
// ponytail: RetryProducer / RegenerateArtifactOnly / EscalateHuman all fold into a retryable
// failure — the existing retryStageOrFail loop re-runs the full producer (incl. the tools-less
// emission nudge) and surfaces exhaustion to the operator. A dedicated regenerate-only /
// approval-pause mid-artifact path is deferred until one is shown to help.
return ArtifactLadderOutcome.Reject(
StageExecutionResult.Failure(
"artifact repair failed (${unresolved.classification}): ${unresolved.detail}",
retryable = decision != ArtifactPolicyDecision.Abort,
),
)
}
/** One tools-less, grammar-constrained inference asking the model to repair its own malformed artifact. */
internal suspend fun SessionOrchestrator.runArtifactRepairInference(
sessionId: SessionId,
stageId: StageId,
slot: TypedArtifactSlot,
bestCandidate: String,
stageConfig: StageConfig,
effectives: RunEffectives,
timeoutMs: Long,
): String? {
val schema = slot.kind.deriveJsonSchema()
val schemaJson = Json.encodeToString(JsonSchema.serializer(), schema)
val prompt = "The previous output for the '${slot.name.value}' artifact was malformed and did not " +
"match the required schema.\n\nMalformed output:\n$bestCandidate\n\nReturn ONLY a single JSON " +
"object matching this schema. Do not add fields, prose, or code fences.\nSchema: $schemaJson"
val entry = ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()),
layer = ContextLayer.L2,
sourceType = "artifactRepair",
sourceId = UUID.randomUUID().toString(),
content = prompt,
tokenEstimate = estimateTokens(prompt),
role = EntryRole.USER,
)
val ctx = contextPackBuilder.build(
id = ContextPackId(UUID.randomUUID().toString()),
sessionId = sessionId,
stageId = stageId,
entries = listOf(entry),
budget = TokenBudget(limit = stageConfig.tokenBudget),
)
val result = runInference(
sessionId, stageId, ctx, stageConfig, timeoutMs, ResponseFormat.Json(schema), effectives, withTools = false,
)
return (result as? InferenceResult.Success)?.response?.text
}
internal suspend fun SessionOrchestrator.emitArtifactRepairAttempted(
sessionId: SessionId,
stageId: StageId,
slot: TypedArtifactSlot,
classification: ArtifactFailure,
rung: String,
) = emit(
sessionId,
ArtifactRepairAttemptedEvent(slot.name, classification.name, rung, sessionId, stageId),
)
internal suspend fun SessionOrchestrator.emitArtifactRepairResolved(
sessionId: SessionId,
stageId: StageId,
slot: TypedArtifactSlot,
json: String,
) {
val hash = artifactStore.put(json.toByteArray())
emit(sessionId, ArtifactRepairResolvedEvent(slot.name, hash, sessionId, stageId))
}
internal fun SessionOrchestrator.stampBuckets(entries: List<ContextEntry>): List<ContextEntry> = entries.map {
if (it.sourceType in REQUIRED_SOURCE_TYPES) it.copy(bucket = ContextBucket.REQUIRED) else it
}
internal fun SessionOrchestrator.parseFileWrittenArtifact(json: String): FileWrittenArtifact? =
runCatching { Json.decodeFromString(FileWrittenArtifact.serializer(), json) }.getOrNull()
/**
* Projects the full change set of the stage(s) that produced a file_written artifact from
* recorded events: ArtifactContentStoredEvent → producing stageIds, ToolInvocationRequested →
* that stage's invocations, FileWrittenEvent → the paths actually written. Pure projection
* over existing events — no new artifact kind, no producer change, replay-safe. Null when no
* writes are on record (caller falls back to the cached single-file JSON).
*/
internal fun SessionOrchestrator.fileWrittenManifest(sessionId: SessionId, needed: ArtifactId): String? {
val events = eventStore.read(sessionId)
val stageIds = events.mapNotNull { it.payload as? ArtifactContentStoredEvent }
.filter { it.artifactId == needed }
.map { it.stageId }
.toSet()
if (stageIds.isEmpty()) return null
val invocationIds = events.mapNotNull { it.payload as? ToolInvocationRequestedEvent }
.filter { it.stageId in stageIds }
.map { it.invocationId }
.toSet()
val paths = events.mapNotNull { it.payload as? FileWrittenEvent }
.filter { it.invocationId in invocationIds && it.postImageHash != null }
.map { it.path }
.distinct()
if (paths.isEmpty()) return null
return buildString {
appendLine("Files written by the producing stage (use file_read to load any content you need):")
paths.forEach { appendLine("- $it") }
}.trimEnd()
}
internal suspend fun SessionOrchestrator.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.
*/
// emit_artifact gives the model a structured channel to produce its required artifact: it calls
// the tool with the fields filled in (parameter schema = the artifact's own schema) instead of
// emitting free-form JSON as a final message. Tool-calling models handle this far more reliably,
// and it sidesteps llama.cpp's grammar+tools incompatibility (the artifact can't be grammar-
// constrained while tools are present). Empty when the stage has no LLM-emitted slot.
internal fun SessionOrchestrator.emitArtifactTool(stageConfig: StageConfig): List<ToolDefinition> {
val slot = stageConfig.produces.firstOrNull { it.kind.llmEmitted } ?: return emptyList()
val schema = Json.encodeToJsonElement(JsonSchema.serializer(), slot.kind.deriveJsonSchema()).jsonObject
return listOf(
ToolDefinition(
function = ToolFunction(
name = EMIT_ARTIFACT_TOOL,
description = "Emit the required '${slot.name.value}' artifact by filling in the fields " +
"below. Prefer this over writing the JSON as a plain message.",
parameters = schema,
),
),
)
}
@@ -0,0 +1,376 @@
package com.correx.core.kernel.orchestration
import com.correx.core.approvals.ApprovalOutcome
import com.correx.core.artifacts.kind.JsonSchema
import com.correx.core.context.model.ContextEntry
import com.correx.core.context.model.ContextLayer
import com.correx.core.context.model.EntryRole
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
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.RepoKnowledgeHit
import com.correx.core.events.events.RepoKnowledgeRetrievedEvent
import com.correx.core.events.events.RepoMapComputedEvent
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.ExecutionPlanLockedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
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.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.inference.ResponseFormat
import com.correx.core.transitions.evaluation.EvaluationContext
import com.correx.core.transitions.graph.StageConfig
import com.correx.core.transitions.graph.WorkflowGraph
import kotlinx.coroutines.CompletableDeferred
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
import kotlin.coroutines.cancellation.CancellationException
/** 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. */
internal fun SessionOrchestrator.evictArtifactContentCache(sessionId: SessionId) {
val prefix = "${sessionId.value}:"
artifactContentCache.keys.removeAll { it.startsWith(prefix) }
}
/** Deterministic extraction/repair ladder for near-miss LLM artifact text (prose-wrapped JSON,
* code fences, trailing commas). Pure — recomputes on replay, records no events. */
internal suspend fun SessionOrchestrator.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,
),
)
}
internal suspend fun SessionOrchestrator.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].
*/
internal suspend fun SessionOrchestrator.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 SessionOrchestrator.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 >= tuning.maxClarificationRounds) 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 the recorded repo map as a structural "what exists" listing — directories with
* their files, alphabetical, no recency ranking and no symbol excerpts. Relevance is the
* retrieval layer's job (buildRelevantFilesEntry); the map only tells the model what it can
* `file_read`. Markdown/docs paths are gated: they only appear when the stage prompt
* explicitly asks about documentation, so churny kernel docs can't bury real source
* (2026-07-05 context-poisoning root cause). Reads the latest [RepoMapComputedEvent] from
* the log (replay-safe — never re-scans the FS).
*/
internal suspend fun SessionOrchestrator.buildRepoMapEntries(sessionId: SessionId, stagePrompt: String = ""): List<ContextEntry> {
val map = eventStore.read(sessionId)
.mapNotNull { it.payload as? RepoMapComputedEvent }
.lastOrNull() ?: return emptyList()
val docsWanted = DOCS_REQUEST_REGEX.containsMatchIn(stagePrompt)
val paths = map.entries.map { it.path }.distinct()
.filter { docsWanted || !isDocPath(it) }
.sorted()
if (paths.isEmpty()) return emptyList()
val byDir = paths.groupBy { it.substringBeforeLast('/', missingDelimiterValue = ".") }
val content = buildString {
appendLine("## Repo layout (what exists — use file_read for content)")
byDir.entries.sortedBy { it.key }.forEach { (dir, files) ->
val names = files.map { it.substringAfterLast('/') }
val shown = names.take(tuning.repoMapFilesPerDir)
val more = names.size - shown.size
val suffix = if (more > 0) ", …(+$more more)" else ""
appendLine("- $dir/ (${names.size}): ${shown.joinToString(", ")}$suffix")
}
}.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,
),
)
}
/**
* True when [stageId]'s own prompt must be suppressed because the stage was entered to repair a
* gate failure. That happens when the most recent transition INTO this stage was a ticket route
* ([TICKET_ROUTE], same entry check as DefaultSessionOrchestrator.ticketReturnMove) — the failure
* ticket routes an owner stage back to fix a file it authored, and its original generative mandate
* ("scaffold the project") is exactly what drives it to overwrite that file with a stub. The
* recovery-ticket entry ([buildRecoveryTicketEntry]) becomes the sole mandate instead.
*
* The purpose-built recovery arbiter (role=recovery) is excluded: it is also ticket-entered but
* its own prompt IS a tuned repair prompt, so it keeps it. Replay-safe (recorded events only).
*/
internal suspend fun SessionOrchestrator.buildContextualRepoEntries(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
): List<ContextEntry> {
val stagePrompt = repoKnowledgeQuery(sessionId, stageId, stageConfig)
val recorded = eventStore.read(sessionId)
.mapNotNull { it.payload as? RepoKnowledgeRetrievedEvent }
.lastOrNull { it.stageId == stageId }
if (recorded != null) {
return repoEntriesOrMapFloor(sessionId, recorded.hits, stagePrompt) + buildDocsCatalogEntry(sessionId)
}
val retriever = repoKnowledgeRetriever
?: return buildRepoMapEntries(sessionId, stagePrompt) + buildDocsCatalogEntry(sessionId)
val hits = runCatching { retriever.retrieve(sessionId, stagePrompt, tuning.repoMapInjectTopK) }
.getOrElse { e ->
if (e is CancellationException) throw e
log.warn("repo-knowledge retrieval failed for stage {}: {}", stageId.value, e.message)
emptyList()
}
// Record the raw retrieval for audit (a degenerate all-zero result is itself the signal that
// the embedder is noop / the index is empty), but only useful hits ever reach the agent.
if (hits.isNotEmpty()) emit(sessionId, RepoKnowledgeRetrievedEvent(sessionId, stageId, stagePrompt, hits))
return repoEntriesOrMapFloor(sessionId, hits, stagePrompt) + buildDocsCatalogEntry(sessionId)
}
/**
* Always-on, capped catalog of the docs that exist, as `path — descriptor` lines (the descriptor
* is the doc's frontmatter/H1/first-line summary recorded on [RepoMapEntry.symbols]). This is the
* read-on-demand doc index: one line per doc lets a stage decide whether to file_read it, so docs
* stay discoverable without the heading-flood the docs gate was added to suppress. Distinct from
* [buildRepoMapEntries]/[repoEntriesOrMapFloor], which still exclude doc *content* from the source
* layout for non-doc prompts. Reads the recorded map (replay-safe — never re-scans the FS).
*/
internal suspend fun SessionOrchestrator.buildDocsCatalogEntry(sessionId: SessionId): List<ContextEntry> {
val map = eventStore.read(sessionId)
.mapNotNull { it.payload as? RepoMapComputedEvent }
.lastOrNull() ?: return emptyList()
val docs = map.entries
.filter { isDocPath(it.path) }
.sortedByDescending { it.score }
.take(tuning.docsCatalogMax)
if (docs.isEmpty()) return emptyList()
val content = buildString {
appendLine("## Docs available (file_read to open — do not assume contents)")
docs.forEach { entry ->
val descriptor = entry.symbols.firstOrNull()?.takeIf { it.isNotBlank() }
appendLine(if (descriptor != null) "- ${entry.path}$descriptor" else "- ${entry.path}")
}
}.trimEnd()
return listOf(
ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()),
layer = ContextLayer.L3,
content = content,
sourceType = "docsCatalog",
sourceId = "docs-catalog",
tokenEstimate = estimateTokens(content),
role = EntryRole.SYSTEM,
),
)
}
/**
* Zero-score hits carry no ranking signal (a noop embedder scores everything 0.0), so injecting
* them just feeds the agent arbitrary files and invites hallucinated paths. Drop them; when nothing
* useful survives, fall back to the deterministic repo map so the stage still has real grounding.
*
* Markdown/docs hits are gated exactly like the repo-map floor (buildRepoMapEntries): with the
* embedder live, semantic similarity happily ranks churny kernel docs (epics, ADRs, QA plans)
* above real source for a code task, re-poisoning the context the floor's gate was added to fix
* (2026-07-05 root cause). They only survive when the stage prompt explicitly asks about docs.
*/
internal suspend fun SessionOrchestrator.repoEntriesOrMapFloor(
sessionId: SessionId,
hits: List<RepoKnowledgeHit>,
stagePrompt: String,
): List<ContextEntry> {
val docsWanted = DOCS_REQUEST_REGEX.containsMatchIn(stagePrompt)
val useful = hits.filter { it.score > 0f && (docsWanted || !isDocPath(it.path)) }
return if (useful.isEmpty()) {
buildRepoMapEntries(sessionId, stagePrompt)
} else {
listOf(buildRelevantFilesEntry(useful))
}
}
/**
* Stamps the required/optional budget bucket by sourceType. REQUIRED = the stage cannot run
* correctly without it (task, constraints, loaded inputs, failure feedback) — never trimmed,
* build fails fast if it alone exceeds the budget. Everything else (repo map, retrieval,
* journal, profiles, vocabulary, transcript) is OPTIONAL and shares the remainder under
* per-type ceilings.
*/
internal suspend fun SessionOrchestrator.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 {
// file_written handoff: hand the next stage the whole producing stage's change set
// as a manifest (known files), not one file's content. The cache value only holds
// the LAST write of the stage, so rendering it inline both misleads (5 of 6 files
// invisible) and wastes budget — content stays lazy behind file_read.
val manifest = parseFileWrittenArtifact(cached)
?.let { fileWrittenManifest(sessionId, needed) }
"## Input artifact: ${needed.value}\n${manifest ?: 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,
)
}
}
@@ -1,175 +1,26 @@
package com.correx.core.kernel.orchestration
import com.correx.core.approvals.ApprovalOutcome
import com.correx.core.approvals.ApprovalStatus
import com.correx.core.approvals.GRANT_LEDGER_SESSION_ID
import com.correx.core.approvals.ProjectIdentity
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.KindContractTable
import com.correx.core.artifacts.kind.KindInference
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.builder.RequiredContextOverflowException
import com.correx.core.context.model.ContextBucket
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.FileWrittenEvent
import com.correx.core.events.events.ContractGateEvaluatedEvent
import com.correx.core.events.events.PlanCompileCheckedEvent
import com.correx.core.events.events.ReviewFindingsRaisedEvent
import com.correx.core.events.events.ReviewVerdict
import com.correx.core.events.events.ContractAssertionResult
import com.correx.core.events.events.ArtifactCreatedEvent
import com.correx.core.events.events.ArtifactRepairAttemptedEvent
import com.correx.core.events.events.ArtifactRepairFailedEvent
import com.correx.core.events.events.ArtifactRepairResolvedEvent
import com.correx.core.events.events.CritiqueFindingsRecordedEvent
import com.correx.core.events.events.CritiqueOutcomeCorrelatedEvent
import com.correx.core.events.events.BriefEchoMismatchEvent
import com.correx.core.events.events.BriefGroundingCheckedEvent
import com.correx.core.events.events.StaticAnalysisCompletedEvent
import com.correx.core.events.events.StaticAnalysisFinding
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.RepoKnowledgeHit
import com.correx.core.events.events.InitialIntentEvent
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.OutsidePathAccessGrantedEvent
import com.correx.core.events.events.RiskAssessedEvent
import com.correx.core.events.events.ExecutionPlanLockedEvent
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.SessionContextProjection
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.sessions.projections.EgressAllowlistProjection
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.inference.ModelCapability
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.ToolCapability
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.BuildExpectation
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.artifact.ArtifactExtractionPipeline
import com.correx.core.validation.artifact.ArtifactFailure
import com.correx.core.validation.model.ValidationContext
import com.correx.core.validation.model.ValidationReport
import com.correx.core.validation.model.ValidationSeverity
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.JsonArray
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
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.Path
import java.nio.file.Paths
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
@@ -0,0 +1,387 @@
package com.correx.core.kernel.orchestration
import com.correx.core.artifacts.kind.KindContractTable
import com.correx.core.artifacts.kind.KindInference
import com.correx.core.events.events.FileWrittenEvent
import com.correx.core.events.events.ContractGateEvaluatedEvent
import com.correx.core.events.events.PlanCompileCheckedEvent
import com.correx.core.events.events.ContractAssertionResult
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.StaticAnalysisCompletedEvent
import com.correx.core.events.events.GroundingReference
import com.correx.core.events.events.ExecutionPlanLockedEvent
import com.correx.core.events.events.StageCheckpointFailedEvent
import com.correx.core.events.events.ToolInvocationRequestedEvent
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.transitions.graph.StageConfig
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
internal fun SessionOrchestrator.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,
gate = "produces",
)
}
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,
gate = "produces",
)
}
/**
* 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.
*/
internal suspend fun SessionOrchestrator.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,
gate = "brief_grounding",
)
}
}
/**
* Brief echo gate (plan-pipeline-addenda §A1): for a stage that opted in (`brief_echo`),
* diff the model's restatement artifact against the original brief. Both are already in the
* artifact content cache (recorded via ArtifactCreatedEvent), so the diff is a pure function
* of recorded data — deterministically recomputable on replay without re-probing any
* environment (invariants #8, #9). Emits [BriefEchoMismatchEvent] on mismatch as a domain
* signal, then fails the stage (retryable) so the pipeline cannot reach plan generation on a
* misread brief.
*
* Brief-source selection among `needs`: prefers the artifact named "analysis"; if absent,
* falls back to the single needs artifact present. When there are multiple needs and none is
* named "analysis", the gate skips (ambiguous brief) and returns Success.
*/
internal suspend fun SessionOrchestrator.checkBriefEcho(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
): StageExecutionResult {
if (stageConfig.metadata["briefEcho"] != "true") return StageExecutionResult.Success(emptyList())
val echoSlot = stageConfig.produces.firstOrNull { it.kind.llmEmitted }
?: return StageExecutionResult.Success(emptyList())
val echo = artifactContentCache["${sessionId.value}:${echoSlot.name.value}"]
?: return StageExecutionResult.Success(emptyList())
val brief = when {
stageConfig.needs.any { it.value == "analysis" } ->
artifactContentCache["${sessionId.value}:analysis"]
stageConfig.needs.size == 1 ->
artifactContentCache["${sessionId.value}:${stageConfig.needs.single().value}"]
else -> null // ambiguous — skip
} ?: return StageExecutionResult.Success(emptyList())
val div = BriefEchoDiff.compare(brief, echo)
return if (div.isMismatch) {
emit(
sessionId,
BriefEchoMismatchEvent(
sessionId = sessionId,
stageId = stageId,
uncoveredRequirements = div.uncoveredRequirements,
hallucinatedFiles = div.hallucinatedFiles,
hallucinatedSymbols = div.hallucinatedSymbols,
),
)
log.warn(
"[Orchestrator] brief echo mismatch session={} stage={} uncovered={} hallucinatedFiles={}",
sessionId.value, stageId.value,
div.uncoveredRequirements.joinToString(", "),
div.hallucinatedFiles.joinToString(", "),
)
StageExecutionResult.Failure(
"stage ${stageId.value} restated the brief unfaithfully — " +
"dropped requirements: [${div.uncoveredRequirements.joinToString("; ")}]; " +
"files not in the brief: [${div.hallucinatedFiles.joinToString(", ")}]. " +
"Restate the brief exactly; do not drop requirements or invent files.",
retryable = true,
gate = "brief_echo",
)
} else {
StageExecutionResult.Success(emptyList())
}
}
/**
* Post-validation stage gates, run in order; the first failure short-circuits and is returned.
* Cheap deterministic checks run first (produces presence, brief grounding, brief echo), then
* the heavier process-spawning static-analysis step last, so it only runs once a stage is
* otherwise sound.
*/
internal suspend fun SessionOrchestrator.runPostStageGates(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
effectives: RunEffectives,
profileCommands: Map<String, String>,
): StageExecutionResult {
val produced = verifyProduces(sessionId, stageId, stageConfig)
if (produced is StageExecutionResult.Failure) return produced
// Stage-level plan checkpoint (C-A2): once a stage has produced its declared artifacts,
// record the checkpoint before the heavier brief/static gates run.
emitStageCheckpoint(sessionId, stageId, stageConfig)
val gates: List<suspend () -> StageExecutionResult> = listOf(
{ groundBriefReferences(sessionId, stageId, stageConfig, effectives) },
{ checkBriefEcho(sessionId, stageId, stageConfig) },
{ runContractGate(sessionId, stageId, stageConfig, effectives) },
{ runPlanCompileGate(sessionId, stageId, stageConfig) },
{ runStaticAnalysis(sessionId, stageId, stageConfig, effectives) },
{ runExecutionGate(sessionId, stageId, stageConfig, effectives, profileCommands) },
{ runReviewGate(sessionId, stageId, stageConfig, effectives) },
)
for (gate in gates) {
val result = gate()
if (result is StageExecutionResult.Failure) return result
}
return StageExecutionResult.Success(emptyList())
}
/**
* Gate 2 — contract (design 2026-07-06-staged-verification-and-review.md §1). For a stage that
* produces file_written artifacts, project the files it actually wrote (Option B: runtime
* manifest), infer each file's kind from its path, derive the [KindContractTable] assertions
* (universal floor for unknown kinds), and evaluate the FS/TEXT ones deterministically. Every
* verdict is recorded in a [ContractGateEvaluatedEvent] (invariant #9), so replay reads it back
* and never re-inspects the filesystem.
*
* A failed assertion fails the stage retryably with the failing assertion ids + evidence fed
* back verbatim — failing-test-name granular hand-back, so the implementer fixes exactly the
* empty/missing/malformed file rather than read-looping. COMPILER assertions are recorded as
* skipped here; the static-analysis command gate is their authoritative check.
*/
/**
* Evaluate the stage's file contract against current filesystem reality and record the verdict
* in a [ContractGateEvaluatedEvent] (invariant #9). Returns the per-assertion results, or null
* when the gate does not apply (no evaluator, no workspace root, no file_written produces, or no
* paths). Shared by the post-stage [runContractGate] and the within-loop remaining-delta refresh
* ([buildRemainingDeltaEntry]) so both read the same contract-minus-reality computation from a
* single evaluation + single event emission.
*/
internal suspend fun SessionOrchestrator.evaluateStageContract(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
effectives: RunEffectives,
): List<ContractAssertionResult>? {
val evaluator = contractAssertionEvaluator ?: return null
val workspaceRoot = effectives.policy?.workspaceRoot ?: return null
if (stageConfig.produces.none { it.kind.id == "file_written" }) return null
// Option A B: the files the stage actually wrote (runtime manifest) unioned with the
// concrete files the plan declared it would produce (expectedFiles). A declared file that
// was never written survives here and fails file_exists — that is the missing-file catch.
val paths = (stageWrittenPaths(sessionId, stageId) + stageConfig.expectedFiles).distinct()
if (paths.isEmpty()) return null
val assertions = paths.flatMap { path ->
KindContractTable.assertionsFor(KindInference.kindFor(path) ?: "", path)
}
val results = assertions.map { a ->
val v = evaluator.evaluate(workspaceRoot, a)
ContractAssertionResult(
assertionId = a.id,
target = a.target,
layer = a.layer.name,
evaluator = a.evaluator.name,
passed = v.passed,
evidence = v.evidence.takeLast(CONTRACT_EVIDENCE_CAP),
)
}
emit(sessionId, ContractGateEvaluatedEvent(sessionId, stageId, results))
return results
}
/** The currently-failing assertions as (target, assertionId, evidence) triples for the checklist. */
internal fun SessionOrchestrator.contractFailureItems(results: List<ContractAssertionResult>): List<Triple<String, String, String>> =
results.filterNot { it.passed }.map { Triple(it.target, it.assertionId, it.evidence) }
internal suspend fun SessionOrchestrator.runContractGate(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
effectives: RunEffectives,
): StageExecutionResult {
val results = evaluateStageContract(sessionId, stageId, stageConfig, effectives)
?: return StageExecutionResult.Success(emptyList())
val failures = results.filterNot { it.passed }
if (failures.isEmpty()) return StageExecutionResult.Success(emptyList())
log.warn(
"[Orchestrator] contract gate failed session={} stage={} assertions={}",
sessionId.value, stageId.value,
failures.joinToString(", ") { "${it.assertionId}(${it.target})" },
)
val detail = failures.joinToString("\n") { "- ${it.assertionId} on ${it.target}: ${it.evidence}" }
return StageExecutionResult.Failure(
"stage ${stageId.value} did not satisfy its file contract. Fix these before review:\n$detail",
retryable = true,
gate = "contract",
)
}
/**
* Plan-compile gate: for a stage that produces an `execution_plan` artifact (the freestyle
* architect), compile its produced plan JSON as a post-stage gate. A compile failure fails the
* stage retryably with the compiler error fed back, so the existing retry-feedback loop hands it
* to the architect and the model self-corrects — rather than the post-planning compiler rejecting
* a broken plan with no re-plan path (which parked the session in ACTIVE forever). The verdict is
* recorded in a [PlanCompileCheckedEvent] (invariant #9) so replay reads it back and never
* re-compiles.
*/
internal suspend fun SessionOrchestrator.runPlanCompileGate(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
): StageExecutionResult {
val check = planCompilationCheck ?: return StageExecutionResult.Success(emptyList())
val planSlot = stageConfig.produces.firstOrNull { it.kind.id == "execution_plan" }
?: return StageExecutionResult.Success(emptyList())
val json = artifactContentCache["${sessionId.value}:${planSlot.name.value}"]
?: return StageExecutionResult.Success(emptyList())
val error = check.check(json, "freestyle-${sessionId.value}")
emit(sessionId, PlanCompileCheckedEvent(sessionId, stageId, passed = error == null, error = error))
if (error == null) return StageExecutionResult.Success(emptyList())
log.warn(
"[Orchestrator] plan-compile gate failed session={} stage={}: {}",
sessionId.value, stageId.value, error,
)
return StageExecutionResult.Failure(
"stage ${stageId.value} produced an execution_plan that does not compile. " +
"Fix the plan before it can be locked:\n$error",
retryable = true,
gate = "plan_compile",
)
}
/**
* The workspace-relative paths this stage actually wrote, projected from events (invariant #1):
* ToolInvocationRequested → this stage's invocation ids, FileWrittenEvent → the paths written
* under them. Pure projection, replay-safe.
*/
/**
* True if the session has committed to a buildable project — it wrote either a real code module
* (a path whose inferred kind carries the `imports_resolve` COMPILER assertion: TS/React/Kotlin
* source) OR a build manifest ([BUILD_MANIFEST_KINDS]: package.json, a Gradle module) that
* declares a toolchain. Read from the recorded FileWritten manifest (invariant #9: replay-safe,
* no filesystem access), it is the run-time signal that warrants the auto build gate — the
* planner's declared kinds (uniformly `file_written`) never reveal this, but the written paths do.
*
* The manifest case is what catches the degenerate scaffold that writes only `package.json` and
* no sources: `npm run build` then fails for lack of an entry, so the gate rejects the empty
* "project" instead of letting it reach COMPLETED. A docs-only plan writes neither, so it is left
* alone.
*/
internal fun SessionOrchestrator.sessionProducedBuildTarget(sessionId: SessionId): Boolean =
eventStore.read(sessionId)
.mapNotNull { it.payload as? FileWrittenEvent }
.filter { it.postImageHash != null }
.map { it.path }
.distinct()
.any { path ->
val kind = KindInference.kindFor(path) ?: return@any false
kind in BUILD_MANIFEST_KINDS ||
KindContractTable.assertionsFor(kind, path).any { it.id == "imports_resolve" }
}
internal fun SessionOrchestrator.stageWrittenPaths(sessionId: SessionId, stageId: StageId): List<String> {
val events = eventStore.read(sessionId)
val invocationIds = events.mapNotNull { it.payload as? ToolInvocationRequestedEvent }
.filter { it.stageId == stageId }
.map { it.invocationId }
.toSet()
return events.mapNotNull { it.payload as? FileWrittenEvent }
.filter { it.invocationId in invocationIds && it.postImageHash != null }
.map { it.path }
.distinct()
}
/**
* Static-first reviewer gate (role-reliability §5): for a stage that declares `static_analysis`
* commands, run them (compiler / detekt / formatters) against its just-produced output in the
* workspace root. Every result is recorded in a [StaticAnalysisCompletedEvent] — an environment
* observation captured at run time (invariant #9), so replay reads it back and never re-runs.
*
* A non-clean command fails the stage retryably with its output fed back verbatim (§2: compiler/
* test output is ground truth, not LLM noise), so the producing stage fixes the issue and only
* static-clean output ever reaches the downstream LLM reviewer — its context thereby excludes
* anything static tools already catch, with no parsing or context plumbing.
*/
@@ -0,0 +1,274 @@
package com.correx.core.kernel.orchestration
import com.correx.core.events.events.ReviewFindingsRaisedEvent
import com.correx.core.events.events.ReviewVerdict
import com.correx.core.events.events.ArtifactCreatedEvent
import com.correx.core.events.events.StaticAnalysisCompletedEvent
import com.correx.core.events.events.StaticAnalysisFinding
import com.correx.core.events.events.ArtifactValidatedEvent
import com.correx.core.events.events.ArtifactValidatingEvent
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.transitions.graph.BuildExpectation
import com.correx.core.transitions.graph.StageConfig
import com.correx.core.validation.model.ValidationReport
import com.correx.core.validation.model.ValidationSeverity
import java.nio.file.Path
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
internal suspend fun SessionOrchestrator.runStaticAnalysis(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
effectives: RunEffectives,
): StageExecutionResult {
if (stageConfig.staticAnalysis.isEmpty()) return StageExecutionResult.Success(emptyList())
val runner = staticAnalysisRunner
val workspaceRoot = effectives.policy?.workspaceRoot
if (runner == null || workspaceRoot == null) {
log.warn(
"[Orchestrator] stage declares static_analysis but no {} is wired — skipping " +
"session={} stage={}",
if (runner == null) "runner" else "workspace root",
sessionId.value, stageId.value,
)
return StageExecutionResult.Success(emptyList())
}
val results = stageConfig.staticAnalysis.map { command ->
val run = runner.run(workspaceRoot, command)
StaticAnalysisFinding(
command = command,
exitCode = run.exitCode,
summary = run.output.takeLast(STATIC_ANALYSIS_SUMMARY_CAP),
) to run.output
}
emit(sessionId, StaticAnalysisCompletedEvent(sessionId, stageId, results.map { it.first }))
val failures = results.filterNot { it.first.clean }
if (failures.isEmpty()) return StageExecutionResult.Success(emptyList())
log.warn(
"[Orchestrator] static analysis failed session={} stage={} commands={}",
sessionId.value, stageId.value,
failures.joinToString(", ") { it.first.command },
)
val detail = failures.joinToString("\n\n") { (finding, output) ->
"$ ${finding.command} (exit ${finding.exitCode})\n${output.takeLast(STATIC_ANALYSIS_FEEDBACK_CAP)}"
}
return StageExecutionResult.Failure(
"stage ${stageId.value} did not pass static analysis. Fix these before review:\n\n$detail",
retryable = true,
gate = "static_analysis",
)
}
/**
* Gate 4 — execution (staged-verification §1/§2). A stage's [StageConfig.buildExpectation]
* (none/module/project/tests) resolves to the bound project profile's typecheck/build/test
* command and runs it as a deterministic build gate. NONE runs nothing (early stages where the
* project isn't yet runnable). A profile missing the alias skips with a warning — the vocabulary
* degrades safely rather than failing when the operator hasn't configured commands. A non-clean
* run fails the stage retryably with the build output fed back (recorded as a
* [StaticAnalysisCompletedEvent], invariant #9, so replay never re-runs it).
*/
internal suspend fun SessionOrchestrator.runExecutionGate(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
effectives: RunEffectives,
profileCommands: Map<String, String>,
): StageExecutionResult {
// An explicit build_expectation resolves directly. Otherwise, if the compiler flagged this
// (terminal) stage as auto-gateable and the session actually wrote a code module, promote to a
// PROJECT build — the LLM planner never sets build_expectation on its file_written stages, so
// this is the only path that gates a scaffold's compilability. Code-ness is judged here (not at
// compile time) from the real FileWritten manifest, so a docs-only plan is left alone.
val expectation: BuildExpectation? = when {
stageConfig.buildExpectation != BuildExpectation.NONE -> stageConfig.buildExpectation
stageConfig.autoBuildGate && sessionProducedBuildTarget(sessionId) -> BuildExpectation.PROJECT
else -> null
}
val alias = expectation?.commandAlias ?: return StageExecutionResult.Success(emptyList())
val runner = staticAnalysisRunner
val workspaceRoot = effectives.policy?.workspaceRoot
if (runner == null || workspaceRoot == null) return StageExecutionResult.Success(emptyList())
val command = profileCommands[alias]
if (command.isNullOrBlank()) {
log.warn(
"[Orchestrator] stage {} needs a {} build gate but project profile has no '{}' " +
"command — skipping execution gate",
stageId.value, expectation, alias,
)
return StageExecutionResult.Success(emptyList())
}
runSetupCommand(sessionId, stageId, workspaceRoot, profileCommands, runner)?.let { return it }
val run = runner.run(workspaceRoot, command)
val finding = StaticAnalysisFinding(command, run.exitCode, run.output.takeLast(STATIC_ANALYSIS_SUMMARY_CAP))
emit(sessionId, StaticAnalysisCompletedEvent(sessionId, stageId, listOf(finding)))
if (finding.clean) return StageExecutionResult.Success(emptyList())
log.warn(
"[Orchestrator] execution gate failed session={} stage={} expectation={} command={}",
sessionId.value, stageId.value, expectation, command,
)
return StageExecutionResult.Failure(
"stage ${stageId.value} did not pass its $expectation build gate. " +
"Fix these before proceeding:\n\n$ $command (exit ${run.exitCode})\n" +
run.output.takeLast(STATIC_ANALYSIS_FEEDBACK_CAP),
retryable = true,
gate = "execution",
)
}
/**
* Run the operator-declared `setup` command (profile alias [SETUP_COMMAND_ALIAS]) before a build
* gate, when one is configured. A build like `tsc && vite build` fails spuriously if the project
* was never prepared (deps uninstalled, codegen not run) — the funnel reads that as a code failure
* and routes into futile repair. The operator names the fix in their profile (`setup = "npm ci"`,
* `"cargo fetch"`, `"./gradlew dependencies"`, …); the kernel stays ecosystem-agnostic. Runs in the
* workspace root, recorded as a [StaticAnalysisCompletedEvent] (invariant #9). Returns a
* [StageExecutionResult.Failure] when setup itself fails (a real, actionable error), else null (no
* setup configured, or it succeeded — proceed to the build).
*/
internal suspend fun SessionOrchestrator.runSetupCommand(
sessionId: SessionId,
stageId: StageId,
workspaceRoot: Path,
profileCommands: Map<String, String>,
runner: StaticAnalysisRunner,
): StageExecutionResult? {
val setup = profileCommands[SETUP_COMMAND_ALIAS]?.takeIf { it.isNotBlank() } ?: return null
log.info(
"[Orchestrator] running setup command before build gate session={} stage={} command='{}'",
sessionId.value, stageId.value, setup,
)
val run = runner.run(workspaceRoot, setup)
val finding = StaticAnalysisFinding(setup, run.exitCode, run.output.takeLast(STATIC_ANALYSIS_SUMMARY_CAP))
emit(sessionId, StaticAnalysisCompletedEvent(sessionId, stageId, listOf(finding)))
if (finding.clean) return null
return StageExecutionResult.Failure(
"stage ${stageId.value} setup command failed before its build gate. " +
"Fix these before proceeding:\n\n$ $setup (exit ${run.exitCode})\n" +
run.output.takeLast(STATIC_ANALYSIS_FEEDBACK_CAP),
retryable = true,
gate = "execution",
)
}
/**
* Gate 3 — semantic (LLM) review (staged-verification § Gate 3). For a stage that opts in
* (`semanticReview`), the injected [SemanticReviewer] reads the files the stage actually wrote
* (the FileWritten manifest — never blind context) and reviews them against the stage objective,
* returning PR-comment-style findings. Runs last, so its input is static-clean (the deterministic
* funnel already passed) and its findings exclude anything the cheap gates catch.
*
* The reviewer is advisory: every finding is recorded in a [ReviewFindingsRaisedEvent] and
* surfaces to the operator regardless of verdict. A [ReviewVerdict.FAIL] blocks the stage
* retryably ONLY when it carries a high-confidence *correctness* finding and the review-block
* budget ([REVIEW_BLOCK_RETRY_CAP]) is not yet spent — so a stuck reviewer can never trap the
* stage; after the budget, findings surface without blocking. Suboptimal/opinion findings never
* block. Recorded as an environment observation (invariant #9); replay never re-runs the reviewer.
*/
internal suspend fun SessionOrchestrator.runReviewGate(
sessionId: SessionId,
stageId: StageId,
stageConfig: StageConfig,
effectives: RunEffectives,
): StageExecutionResult {
if (!stageConfig.semanticReview) return StageExecutionResult.Success(emptyList())
val reviewer = semanticReviewer ?: return StageExecutionResult.Success(emptyList())
val workspaceRoot = effectives.policy?.workspaceRoot ?: return StageExecutionResult.Success(emptyList())
val files = stageWrittenPaths(sessionId, stageId)
if (files.isEmpty()) return StageExecutionResult.Success(emptyList())
val objective = stageConfig.needs
.mapNotNull { artifactContentCache["${sessionId.value}:${it.value}"] }
.joinToString("\n\n")
.ifBlank { "Stage '${stageId.value}' produced the files below; review them for defects." }
.take(REVIEW_OBJECTIVE_CAP)
val outcome = reviewer.review(sessionId, stageId, workspaceRoot, files, objective)
// A FAIL blocks only on a high-confidence correctness finding, and only while the review-block
// budget remains — count prior review blocks for this stage from the log (replay-safe).
val priorBlocks = eventStore.read(sessionId)
.mapNotNull { it.payload as? ReviewFindingsRaisedEvent }
.count { it.stageId == stageId && it.blocked }
val blockingFinding = outcome.findings.firstOrNull {
it.correctness && it.confidence >= tuning.reviewBlockMinConfidence
}
val shouldBlock = outcome.verdict == ReviewVerdict.FAIL &&
blockingFinding != null &&
priorBlocks < tuning.reviewBlockRetryCap
emit(sessionId, ReviewFindingsRaisedEvent(sessionId, stageId, outcome.verdict, outcome.findings, blocked = shouldBlock))
if (!shouldBlock) {
if (outcome.findings.isNotEmpty()) {
log.info(
"[Orchestrator] semantic review advisory session={} stage={} verdict={} findings={}",
sessionId.value, stageId.value, outcome.verdict, outcome.findings.size,
)
}
return StageExecutionResult.Success(emptyList())
}
log.warn(
"[Orchestrator] semantic review blocked session={} stage={} finding={}",
sessionId.value, stageId.value, blockingFinding?.target,
)
val detail = outcome.findings
.filter { it.correctness }
.joinToString("\n") { "- [${it.severity}] ${it.target}: ${it.message}" +
(it.suggestedFix?.let { fix -> "\n fix: $fix" } ?: "") }
return StageExecutionResult.Failure(
"stage ${stageId.value} failed semantic review — fix these correctness issues:\n$detail",
retryable = true,
gate = "review",
)
}
internal suspend fun SessionOrchestrator.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 ->
emitAll(
sessionId,
listOf(
ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1),
ArtifactValidatingEvent(slot.name, sessionId, stageId),
ArtifactValidatedEvent(slot.name, sessionId, stageId),
),
)
}
}
// Collapse a validation report into a concise, actionable reason. This string becomes the
// RetryAttemptedEvent.failureReason that the retry-feedback entry feeds back to the model, so a
// weak model is told *what* was wrong (e.g. "required property 'summary' missing") instead of a
// bare "validation failed" it can't act on. Capped to keep the feedback prompt small.
internal fun SessionOrchestrator.summarizeRejection(report: ValidationReport): String {
val errors = report.sections.flatMap { section ->
section.issues
.filter { it.severity == ValidationSeverity.ERROR }
.map { it.message }
}.distinct()
return if (errors.isEmpty()) {
"validation failed"
} else {
"validation failed: " + errors.take(tuning.maxFeedbackIssues).joinToString("; ")
}
}
@@ -0,0 +1,292 @@
package com.correx.core.kernel.orchestration
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.withContext
import kotlinx.serialization.json.Json
import kotlinx.serialization.json.JsonArray
import kotlinx.serialization.json.JsonObject
import kotlinx.serialization.json.JsonPrimitive
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.Paths
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
/**
* 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].
*/
internal suspend fun computeToolPreview(
toolName: String,
parameters: Map<String, Any>,
workspaceRoot: java.nio.file.Path?,
): String? {
if (toolName == "shell") return shellCommandPreview(parameters)
if (toolName == "task_decompose") return renderDecomposePreview(parameters)
if (toolName == "file_edit") return computeFileEditPreview(parameters, workspaceRoot)
if (toolName != "file_write") return null
val path = parameters["path"] as? String ?: return null
// file_write no longer carries an `operation` param (delete was split into file_delete), so the
// preview is unconditional: path + content is always a write. The old `operation == "write"` gate
// silently bailed to the raw-JSON args fallback for every file_write.
val proposedContent = parameters["content"] as? String ?: return null
val existingContent = readFileIfExists(path, workspaceRoot)
return buildDiffString(path, existingContent, proposedContent)
}
internal suspend fun readFileIfExists(path: String, workspaceRoot: java.nio.file.Path?): String? =
withContext(Dispatchers.IO) {
runCatching {
// Resolve relative paths against the session's workspace root, same as the tools do —
// resolving against the daemon CWD showed the operator the wrong file (or nothing) when
// server CWD ≠ workspace_root.
val raw = java.nio.file.Paths.get(path)
val filePath = if (raw.isAbsolute || workspaceRoot == null) raw else workspaceRoot.resolve(raw)
if (java.nio.file.Files.exists(filePath)) {
java.nio.file.Files.readString(filePath)
} else null
}.getOrNull()
}
/**
* Preview for `file_edit`'s "replace" and "append" operations: reads the existing file and
* simulates the edit locally to produce a diff, mirroring `file_write`'s preview. "patch" is left
* to the raw-JSON-args fallback — simulating `patch -p1` application here isn't worth the risk of
* the preview silently disagreeing with what the tool actually does.
*/
internal suspend fun computeFileEditPreview(
parameters: Map<String, Any>,
workspaceRoot: java.nio.file.Path?,
): String? {
val path = parameters["path"] as? String ?: return null
val operation = parameters["operation"] as? String ?: return null
val existingContent = readFileIfExists(path, workspaceRoot) ?: return null
val proposedContent = when (operation) {
"append" -> existingContent + (parameters["content"] as? String ?: return null)
"replace" -> {
val target = parameters["target"] as? String ?: return null
val replacement = parameters["replacement"] as? String ?: return null
if (!existingContent.contains(target)) return null
existingContent.replaceFirst(target, replacement)
}
else -> return null
}
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.
*/
internal 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. */
internal fun shellQuoteToken(token: String): String {
if (token.isEmpty()) return "''"
val needsQuote = token.any { it.isWhitespace() || it in "\"'\\|&;<>(){}\$`*?[]~#!" }
return if (needsQuote) "'" + token.replace("'", "'\\''") + "'" else token
}
internal enum class DiffOpKind { EQUAL, DELETE, INSERT }
internal data class DiffOp(val kind: DiffOpKind, val line: String)
/**
* Build a unified-diff string for a full file replacement, using a line-level LCS diff
* so a small edit to a large file shows as a tight hunk rather than a full-file replacement.
* When the file doesn't exist yet (new file), only `+` lines are shown.
*/
internal fun buildDiffString(path: String, existingContent: String?, proposedContent: String): String {
val existingLines = existingContent?.lines() ?: emptyList()
val proposedLines = proposedContent.lines()
if (existingLines == proposedLines) return " (no change)"
val ops = diffLines(existingLines, proposedLines)
return buildString {
appendLine("--- a/$path")
appendLine("+++ b/$path")
append(renderHunks(ops))
}.trimEnd()
}
/** Line-level LCS diff producing an edit script of EQUAL/DELETE/INSERT ops. */
internal fun diffLines(oldLines: List<String>, newLines: List<String>): List<DiffOp> {
val n = oldLines.size
val m = newLines.size
val lcs = Array(n + 1) { IntArray(m + 1) }
for (i in n - 1 downTo 0) {
for (j in m - 1 downTo 0) {
lcs[i][j] = if (oldLines[i] == newLines[j]) {
lcs[i + 1][j + 1] + 1
} else {
maxOf(lcs[i + 1][j], lcs[i][j + 1])
}
}
}
val ops = mutableListOf<DiffOp>()
var i = 0
var j = 0
while (i < n && j < m) {
when {
oldLines[i] == newLines[j] -> {
ops += DiffOp(DiffOpKind.EQUAL, oldLines[i])
i++
j++
}
lcs[i + 1][j] >= lcs[i][j + 1] -> {
ops += DiffOp(DiffOpKind.DELETE, oldLines[i])
i++
}
else -> {
ops += DiffOp(DiffOpKind.INSERT, newLines[j])
j++
}
}
}
while (i < n) {
ops += DiffOp(DiffOpKind.DELETE, oldLines[i])
i++
}
while (j < m) {
ops += DiffOp(DiffOpKind.INSERT, newLines[j])
j++
}
return ops
}
/** Groups an edit script into unified-diff hunks with [context] lines of surrounding EQUAL context. */
internal fun renderHunks(ops: List<DiffOp>, context: Int = 3): String {
val changedIndices = ops.indices.filter { ops[it].kind != DiffOpKind.EQUAL }
if (changedIndices.isEmpty()) return ""
data class Hunk(val start: Int, val end: Int)
val hunks = mutableListOf<Hunk>()
var hunkStart = changedIndices.first()
var hunkEnd = changedIndices.first()
for (idx in changedIndices.drop(1)) {
if (idx - hunkEnd <= context * 2) {
hunkEnd = idx
} else {
hunks += Hunk(hunkStart, hunkEnd)
hunkStart = idx
hunkEnd = idx
}
}
hunks += Hunk(hunkStart, hunkEnd)
var oldLine = 1
var newLine = 1
var opIdx = 0
return buildString {
for (hunk in hunks) {
val from = maxOf(0, hunk.start - context)
val to = minOf(ops.size - 1, hunk.end + context)
while (opIdx < from) {
if (ops[opIdx].kind != DiffOpKind.INSERT) oldLine++
if (ops[opIdx].kind != DiffOpKind.DELETE) newLine++
opIdx++
}
val oldStart = oldLine
val newStart = newLine
var oldCount = 0
var newCount = 0
val lines = StringBuilder()
for (k in from..to) {
val op = ops[k]
when (op.kind) {
DiffOpKind.EQUAL -> {
lines.append(' ').append(op.line).append('\n')
oldCount++
newCount++
}
DiffOpKind.DELETE -> {
lines.append('-').append(op.line).append('\n')
oldCount++
}
DiffOpKind.INSERT -> {
lines.append('+').append(op.line).append('\n')
newCount++
}
}
}
appendLine("@@ -$oldStart,${oldCount.coerceAtLeast(1)} +$newStart,${newCount.coerceAtLeast(1)} @@")
append(lines)
oldLine = oldStart + oldCount
newLine = newStart + newCount
opIdx = to + 1
}
}
}
/**
* The candidate in [paths] that shares [guessed]'s filename and the most trailing path segments,
* or null when nothing shares the filename. Lets a wrong package/dir prefix
* (`com/correx/server/routes/SessionRoutes.kt`) still resolve to the real file
* (`.../com/correx/apps/server/routes/SessionRoutes.kt`) without ever inventing a non-existent path.
*/
internal fun closestPathByFilename(guessed: String, paths: List<String>): String? {
val guessedName = guessed.substringAfterLast('/')
if (guessedName.isBlank()) return null
val guessedSegs = guessed.split('/').filter { it.isNotBlank() }
return paths.asSequence()
.filter { it.substringAfterLast('/') == guessedName }
.maxByOrNull { candidate ->
val segs = candidate.split('/').filter { it.isNotBlank() }
var shared = 0
var i = segs.size - 1
var j = guessedSegs.size - 1
while (i >= 0 && j >= 0 && segs[i] == guessedSegs[j]) {
shared++; i--; j--
}
shared
}
}
/**
* Human-readable approval-card preview of a `task_decompose` call: the proposed epic and the numbered
* child tasks with their "after" (dependency) labels — instead of the truncated raw JSON the generic
* fallback would show. The nested `tasks`/`parent` args arrive as JSON text (flattened), so they are
* re-parsed here. Null on any parse miss, so the caller falls back to the raw arguments.
*/
internal fun renderDecomposePreview(parameters: Map<String, Any>): String? {
val project = parameters["project"] as? String ?: return null
val tasks = (parameters["tasks"] as? String)
?.let { runCatching { Json.parseToJsonElement(it) }.getOrNull() } as? JsonArray ?: return null
if (tasks.isEmpty()) return null
val titleAt = { i: Int -> ((tasks.getOrNull(i) as? JsonObject)?.get("title") as? JsonPrimitive)?.content }
val refToIndex = tasks.mapIndexedNotNull { i, e ->
((e as? JsonObject)?.get("ref") as? JsonPrimitive)?.content?.let { it to i }
}.toMap()
val parentTitle = (parameters["parent"] as? String)
?.let { runCatching { Json.parseToJsonElement(it) }.getOrNull() as? JsonObject }
?.let { (it["title"] as? JsonPrimitive)?.content }
return buildString {
append("Decompose '").append(project).append("' into:")
parentTitle?.let { append("\n epic: ").append(it) }
tasks.forEachIndexed { i, e ->
val o = e as? JsonObject
append("\n ").append(i + 1).append(". ").append((o?.get("title") as? JsonPrimitive)?.content ?: "(untitled)")
val afters = ((o?.get("depends_on") as? JsonArray)?.mapNotNull { (it as? JsonPrimitive)?.content } ?: emptyList())
.map { d -> (refToIndex[d] ?: d.toIntOrNull())?.let { titleAt(it) } ?: d }
if (afters.isNotEmpty()) append(" (after: ").append(afters.joinToString(", ")).append(")")
}
}
}
@@ -0,0 +1,91 @@
package com.correx.core.kernel.orchestration
import com.correx.core.context.model.ContextPack
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.ContextTruncatedEvent
import com.correx.core.events.events.InitialIntentEvent
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.transitions.graph.StageConfig
import kotlinx.datetime.Clock
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
internal fun SessionOrchestrator.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.
*/
internal fun SessionOrchestrator.resolveStagePromptText(stageId: StageId, stageConfig: StageConfig): String {
stageConfig.metadata["promptInline"]?.trim()?.takeIf { it.isNotBlank() }?.let { return it }
stageConfig.metadata["prompt"]?.let { path ->
runCatching { promptResolver.resolve(path) }.getOrNull()
?.trim()?.takeIf { it.isNotBlank() }
?.let { return it }
}
return stageId.value
}
/**
* The role's own prompt text (see [resolveStagePromptText]) is identical across every session
* that runs the stage — it carries zero session-specific signal, so similarity against it can
* only ever reflect generic lexical overlap with the role's boilerplate ("file_read", "ls",
* "grep"...), never what the user actually asked for. The one thing that IS session-specific
* is the initial user intent, so it's prepended here as the real anchor for the query (2026-07-08
* finding: without it, the query matched a tool script's own symbol names over the files the
* user's request actually named).
*/
internal fun SessionOrchestrator.repoKnowledgeQuery(sessionId: SessionId, stageId: StageId, stageConfig: StageConfig): String {
val intent = eventStore.read(sessionId)
.mapNotNull { it.payload as? InitialIntentEvent }
.firstOrNull()?.intent?.trim()
val stagePrompt = resolveStagePromptText(stageId, stageConfig)
return if (intent.isNullOrBlank()) stagePrompt else "$intent\n\n$stagePrompt"
}
// --- 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.
internal suspend fun SessionOrchestrator.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 fun SessionOrchestrator.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.
*/
@@ -0,0 +1,732 @@
package com.correx.core.kernel.orchestration
import com.correx.core.approvals.ApprovalStatus
import com.correx.core.approvals.GRANT_LEDGER_SESSION_ID
import com.correx.core.approvals.ProjectIdentity
import com.correx.core.approvals.Tier
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.kind.FileWrittenArtifact
import com.correx.core.artifacts.kind.ProcessResultArtifact
import com.correx.core.artifacts.kind.TypedArtifactSlot
import com.correx.core.context.model.ContextEntry
import com.correx.core.context.model.ContextLayer
import com.correx.core.context.model.EntryRole
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.ArtifactContentStoredEvent
import com.correx.core.events.events.FileWrittenEvent
import com.correx.core.events.events.ArtifactCreatedEvent
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.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.OutsidePathAccessGrantedEvent
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.toAssessedIssues
import com.correx.core.toolintent.toRiskSummary
import com.correx.core.sessions.projections.EgressAllowlistProjection
import com.correx.core.events.types.ApprovalRequestId
import com.correx.core.events.types.ContextEntryId
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.ToolCallRequest
import com.correx.core.sessions.ApprovalMode
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.ToolCapability
import com.correx.core.tools.contract.ToolResult
import com.correx.core.transitions.graph.StageConfig
import kotlinx.coroutines.CompletableDeferred
import kotlinx.datetime.Clock
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.Paths
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
internal fun SessionOrchestrator.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()
}
// On a recoverable tool failure, append the tool's argument schema to the error fed back to the
// model so it can self-correct its next attempt (it already sees its own malformed call in the
// assistant entry above) rather than repeating the same mistake. Kept compact to bound context.
// NOTE: the "corrected arguments" phrasing was removed because it actively drives loops: a model
// that called list_dir frontend/ and got "directory not found" interprets "re-issue with corrected
// arguments" as "retry the same call" rather than "try a different path" — the arguments were
// valid, the path simply doesn't exist. The schema reference stays so malformed calls can still
// self-correct; the model must infer the tactic from the tool's error text.
internal fun SessionOrchestrator.toolArgsHint(tool: Tool?): String =
tool?.let {
"\nAccepted parameters for '${it.name}': ${it.parametersSchema}"
}.orEmpty()
/**
* Render a tool result into the consistently-framed, bounded text the model sees. Success output
* is run through the tool's [ToolOutputCompressor], then framed with a uniform `[tool exit=N]`
* header. If it still exceeds [TOOL_RESULT_MAX_CHARS] the FULL raw output is spilled to the
* artifact store (CAS — durable, its hash recorded on the ToolReceipt in the event log) and only a
* head+tail preview is shown, with a marker telling the model to call `tool_output(ref=…)` for the
* rest. Failures keep their `ERROR:`/`FATAL:` sentinels unchanged — the all-rejected loop breaker
* keys on those prefixes.
*/
internal suspend fun SessionOrchestrator.renderToolResult(toolName: String, tool: Tool?, result: ToolResult): RenderedToolResult =
when (result) {
is ToolResult.Failure -> RenderedToolResult(
if (!result.recoverable) "FATAL: ${result.reason}" else "ERROR: ${result.reason}${toolArgsHint(tool)}",
null,
)
is ToolResult.Success -> {
val compressed = tool?.outputCompressor?.compress(result.output, ToolOutputContext(result.exitCode))
?: result.output
val header = "[$toolName exit=${result.exitCode}]"
if (compressed.length <= TOOL_RESULT_MAX_CHARS) {
RenderedToolResult("$header\n$compressed", null)
} else {
// Spill the full RAW output (most complete) so retrieval returns everything, not the
// already-compressed preview the model saw.
val ref = artifactStore.put(result.output.toByteArray()).value
RenderedToolResult(frameTruncatedToolResult(header, compressed, ref), ref)
}
}
}
internal suspend fun SessionOrchestrator.dispatchToolCalls(
sessionId: SessionId,
stageId: StageId,
toolCalls: List<ToolCallRequest>,
stageConfig: StageConfig,
effectives: RunEffectives,
approvalMode: ApprovalMode,
// The reasoning that preceded this whole batch of tool calls (they all came from one
// inference response). Attached only to the first call's assistant entry — one thought
// block per model turn, matching how the response actually arrived.
reasoning: String? = null,
): 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.withIndex().flatMap { (toolCallIndex, toolCall) ->
// Only the first call's assistant entry carries the reasoning — the whole batch came
// from one model turn/one thought block, so later calls in the same batch get none.
val toolCallReasoning = reasoning.takeIf { toolCallIndex == 0 }
val invocationId = ToolInvocationId(UUID.randomUUID().toString())
val parameters = parseToolArguments(toolCall.function.arguments)
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
// Out-of-workspace read access: a read whose target lexically resolves outside the
// workspace root. The intent plane raises PROMPT_USER for it; once the operator approves
// (recorded as OutsidePathAccessGrantedEvent), the same path this session skips the prompt
// and the tool jail is widened for it. Writes never take this path (isRead gate).
val isRead = tool?.requiredCapabilities?.contains(ToolCapability.FILE_READ) == true &&
tool.requiredCapabilities.contains(ToolCapability.FILE_WRITE) != true
val outsideReadTarget = if (isRead) {
outsideWorkspaceTarget(parameters, effectives.policy?.workspaceRoot)
} else {
null
}
val grantedOutside = grantedOutsidePaths(sessionId)
val alreadyGranted = outsideReadTarget != null && outsideReadTarget.toString() in grantedOutside
emit(
sessionId,
ToolInvocationRequestedEvent(
invocationId = invocationId,
sessionId = sessionId,
stageId = stageId,
toolName = toolCall.function.name,
tier = tier,
request = request,
capabilities = tool?.requiredCapabilities ?: emptySet(),
),
)
val deniedByStage = toolCall.function.name != STAGE_COMPLETE_TOOL &&
toolCall.function.name !in stageConfig.effectiveAllowedTools
val deniedByReadOnly = !deniedByStage && isReadOnlyMode(sessionId) &&
(tool?.requiredCapabilities?.contains(ToolCapability.FILE_WRITE) == true)
if (deniedByStage || deniedByReadOnly) {
val denyReason = if (deniedByReadOnly) {
"read-before-write: '${toolCall.function.name}' requires reading the file first — use file_read"
} else {
"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,
reasoning = toolCallReasoning,
),
ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()),
layer = ContextLayer.L2,
sourceType = "toolResult",
sourceId = sourceId,
content = "ERROR: $denyReason",
tokenEstimate = estimateTokens(denyReason),
role = EntryRole.TOOL,
),
)
}
// Per-task write scope: while a task is claimed, narrow the manifest to its affected
// paths (recorded on the task) so the implementer can't write outside its unit of work.
// Falls back to the stage's static manifest when nothing is claimed.
val effectiveManifest = taskClaimCoordinator?.activeScope(sessionId)?.takeIf { it.isNotEmpty() }
?: stageConfig.writeManifest
val plane2Risk: RiskSummary? = runPlane2Assessment(
sessionId, stageId, invocationId, toolCall.function.name, request, tool, effectives,
effectiveManifest,
)?.let { assessment ->
if (assessment.recommendedAction == RiskAction.BLOCK) {
val rationale = assessment.rationale.joinToString("; ")
// On a bad-path block, point the model at the closest real file so it fixes the
// path instead of retrying the same wrong guess (deep package paths are easy to
// misremember). Only fires when we can name a concrete match from the repo map.
val suggestion = if (assessment.rationale.any { it.contains(REFERENCE_EXISTS_CODE) }) {
(parameters["path"] as? String)
?.let { suggestClosestPath(sessionId, it) }
?.let { " Did you mean: $it ?" }
.orEmpty()
} else {
""
}
emit(
sessionId,
ToolExecutionRejectedEvent(
invocationId = invocationId,
sessionId = sessionId,
toolName = toolCall.function.name,
tier = tier,
// Record the specific rule rationale, not a generic string, so the audit
// trail (and task history) shows why a call was blocked.
reason = rationale.ifBlank { "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,
reasoning = toolCallReasoning,
),
ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()),
layer = ContextLayer.L2,
sourceType = "toolResult",
sourceId = sourceId,
content = "BLOCKED: $rationale$suggestion",
tokenEstimate = estimateTokens(rationale + suggestion),
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) || alreadyGranted) {
// no approval needed — either within the auto-approve tier, or this out-of-workspace
// read path was already approved earlier this session (this-path-this-session).
} else {
// Grants in effect = this session's own (SESSION/STAGE) unioned with the
// cross-session ledger (PROJECT/GLOBAL). projectId is derived from the bound
// workspace root so PROJECT grants match later sessions on the same repo.
val sessionGrants = approvalRepository.getApprovalState(sessionId).grants.values
val ledgerGrants = approvalRepository.getApprovalState(GRANT_LEDGER_SESSION_ID).grants.values
val activeGrants = (sessionGrants + ledgerGrants).toList()
val projectId = effectives.policy?.workspaceRoot?.let { ProjectIdentity.of(it.toString()) }
val approvalCtx = ApprovalContext(
identity = ApprovalScopeIdentity(sessionId, stageId, projectId = projectId),
mode = approvalMode,
)
val requestId = ApprovalRequestId(UUID.randomUUID().toString())
val toolPreview = computeToolPreview(
toolCall.function.name, parameters, effectives.policy?.workspaceRoot,
)
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"
blockTaskOnScopeRejection(sessionId, toolCall.function.name, rejectReason)
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,
reasoning = toolCallReasoning,
),
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"
blockTaskOnScopeRejection(sessionId, toolCall.function.name, rejectReason)
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,
reasoning = toolCallReasoning,
),
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))
}
}
// Reaching here means the call is authorized (all deny/block/pending paths returned early).
// For a first-time out-of-workspace read, record the grant so future reads of the same path
// this session skip the prompt (invariants #8/#9 — replay reads the recorded fact). Then
// widen the executed request's jail with every path approved outside the workspace.
if (outsideReadTarget != null && !alreadyGranted) {
emit(
sessionId,
OutsidePathAccessGrantedEvent(
sessionId = sessionId,
stageId = stageId,
path = outsideReadTarget.toString(),
),
)
}
val executedRequest = outsideReadTarget
?.let { request.copy(grantedPaths = grantedOutside + it.toString()) }
?: request
val result = executor.execute(executedRequest)
// Frame + bound the result once; on truncation this spills the full output to CAS and
// returns its hash, which we record on the receipt so the event log points at the full text.
val rendered = renderToolResult(toolCall.function.name, tool, result)
// 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, rendered.fullOutputHash,
)
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,
reasoning = toolCallReasoning,
)
val resultEntry = ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()),
layer = ContextLayer.L2,
sourceType = "toolResult",
sourceId = sourceId,
content = rendered.content,
tokenEstimate = estimateTokens(rendered.content),
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)
}
}
/**
* A rejected [SCOPE_PROPOSAL_TOOL] call means the operator denied widening the claimed task's
* scope — block the task so the loop advances instead of the implementer retrying the same
* out-of-scope write. Only fires for the scope-proposal tool; other denied tools are unaffected.
*/
internal suspend fun SessionOrchestrator.blockTaskOnScopeRejection(sessionId: SessionId, toolName: String, reason: String) {
if (toolName == SCOPE_PROPOSAL_TOOL) {
taskClaimCoordinator?.blockActiveTask(sessionId, "scope amendment rejected: $reason")
}
}
internal suspend fun SessionOrchestrator.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
// Resolve the egress hosts granted to this session by folding its EgressHostsGrantedEvents
// through the projection. Unioned with the static allow-list in NetworkHostRule; empty when
// nothing has been granted, which never narrows the static allow-list.
val sessionEgressHosts = resolveSessionEgressHosts(sessionId)
val assessment = assessor.assess(
ToolCallAssessmentInput(
request = request,
capabilities = tool?.requiredCapabilities ?: emptySet(),
workspace = policy,
probe = worldProbe,
paramRoles = tool?.paramRoles ?: emptyMap(),
writeManifest = writeManifest,
sessionEgressHosts = sessionEgressHosts,
session = resolveSessionContext(sessionId),
),
)
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()
}
/**
* Folds this session's [com.correx.core.events.events.EgressHostsGrantedEvent]s through
* [EgressAllowlistProjection] into the running union of hosts granted to it. Replay-safe (reads
* the log, never live state); empty when no grants have been emitted.
*/
internal suspend fun SessionOrchestrator.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
}
emitAll(
sessionId,
listOf(
ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1),
ArtifactValidatingEvent(slot.name, sessionId, stageId),
ArtifactValidatedEvent(slot.name, sessionId, stageId),
),
)
}
}
internal suspend fun SessionOrchestrator.recordToolExecution(
sessionId: SessionId,
stageId: StageId,
toolCall: ToolCallRequest,
invocationId: ToolInvocationId,
tier: Tier,
result: ToolResult,
fileTool: FileAffectingTool?,
request: ToolRequest,
fileWrittenSlots: List<TypedArtifactSlot>,
fullOutputHash: String? = null,
) {
// Invariant #5: every tool side effect is captured. This is the single authoritative
// ToolExecutionCompleted/Failed record and the source the read-before-write gate replays.
// FileWrittenEvent (with pre/post-image hashes for reversibility) and ToolExecutionStarted are
// emitted once, by SandboxedToolExecutor — NOT here — so completions aren't duplicated in the log.
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()
// Read-only tools (file_read, list_dir) have no FileAffectingTool, so `affected`
// is empty — fall back to the raw `path` argument so the receipt still names
// the file/dir the tool looked at (surfaced in the TUI's output row).
val affectedNames = affected.map { it.toString() }
.ifEmpty { listOfNotNull(request.parameters["path"] as? String) }
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),
// Carry the tool's structured metadata (e.g. file_read's contentHash) so
// session projections can read it.
structuredOutput = result.metadata,
affectedEntities = affectedNames,
durationMs = 0,
tier = tier,
timestamp = Clock.System.now(),
diff = diff,
fullOutputHash = fullOutputHash,
),
),
)
if (affected.isNotEmpty()) {
materializeFileWritten(sessionId, stageId, request, affected, fileWrittenSlots, diff)
}
}
}
}
/**
* 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.
*/
internal suspend fun SessionOrchestrator.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,
),
)
}
}
internal suspend fun SessionOrchestrator.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
}
emitAll(
sessionId,
listOf(
ArtifactCreatedEvent(slot.name, sessionId, stageId, schemaVersion = 1),
ArtifactValidatingEvent(slot.name, sessionId, stageId),
ArtifactValidatedEvent(slot.name, sessionId, stageId),
),
)
}
}
@@ -0,0 +1,198 @@
package com.correx.core.kernel.orchestration
import com.correx.core.artifacts.ArtifactState
import com.correx.core.events.events.CritiqueFindingsRecordedEvent
import com.correx.core.events.events.CritiqueOutcomeCorrelatedEvent
import com.correx.core.events.events.EventMetadata
import com.correx.core.events.events.EventPayload
import com.correx.core.events.events.NewEvent
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.types.ArtifactId
import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.kernel.execution.WorkflowResult
import com.correx.core.transitions.evaluation.EvaluationContext
import com.correx.core.transitions.graph.WorkflowGraph
import com.correx.core.transitions.resolution.TransitionDecision
import kotlinx.datetime.Clock
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
// 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.
internal fun SessionOrchestrator.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 SessionOrchestrator.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,
readyTaskCount = readyTaskCounter?.count(sessionId) ?: 0,
)
return transitionResolver.resolve(graph, ctx)
}
internal fun SessionOrchestrator.isTerminal(graph: WorkflowGraph, stageId: StageId): Boolean =
graph.transitions.none { it.from == stageId }
internal suspend fun SessionOrchestrator.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 SessionOrchestrator.emitWorkflowStarted(sessionId: SessionId, graph: WorkflowGraph, config: OrchestrationConfig) {
emit(
sessionId,
WorkflowStartedEvent(
sessionId = sessionId,
workflowId = graph.id,
startStageId = graph.start,
retryPolicy = config.retryPolicy,
),
)
}
internal suspend fun SessionOrchestrator.completeWorkflow(
sessionId: SessionId,
terminalStageId: StageId,
stageCount: Int,
workflowId: String,
): WorkflowResult.Completed {
log.debug(
"[Orchestrator] COMPLETED session={} terminalStage={} stages={}",
sessionId.value, terminalStageId.value, stageCount,
)
correlateCritiqueOutcomes(sessionId)
emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId))
cancellations.remove(sessionId)
evictArtifactContentCache(sessionId)
return WorkflowResult.Completed(sessionId, terminalStageId)
}
internal suspend fun SessionOrchestrator.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,
)
// failWorkflow IS the clean-termination path: whatever goes wrong while recording it
// (event-store hiccup, correlation lookup, or any other transient failure) must never be
// allowed to escape as a raw exception — that would blow past this function's own
// WorkflowFailedEvent (correct stageId/reason/retryExhausted) and land in the server's
// top-level catch-all, which fabricates its own event with the wrong stage (graph.start)
// and a garbage reason (the escaping throwable's message/toString). Record what we can and
// still return a clean Failed result either way.
// Critique correlation is best-effort bookkeeping; if it throws we log and press on, because
// the terminal WorkflowFailedEvent below MUST still be recorded — skipping it would leave the
// event log with no terminal marker for the session (breaks replay-completeness, invariants
// #1/#8) and let the server catch-all fabricate a wrong one.
runCatching { correlateCritiqueOutcomes(sessionId) }.onFailure { e ->
log.error(
"[Orchestrator] failWorkflow: critique correlation failed session={} stage={}",
sessionId.value, stageId.value, e,
)
}
// Last-ditch: the one unrecoverable case. If emitting the terminal event itself throws
// (e.g. a serialization edge), we cannot record the failure at all — log it loudly with the
// full throwable so it is never silent, then still return a clean Failed result.
runCatching {
emit(sessionId, WorkflowFailedEvent(sessionId, stageId, reason, retryExhausted))
}.onFailure { e ->
log.error(
"[Orchestrator] failWorkflow: FAILED to record terminal WorkflowFailedEvent — event " +
"log has no terminal marker for session={} stage={}",
sessionId.value, stageId.value, e,
)
}
cancellations.remove(sessionId)
evictArtifactContentCache(sessionId)
return WorkflowResult.Failed(sessionId, reason, retryExhausted)
}
/**
* At loop resolution, correlate any recorded critique findings (§B-§6) into outcome events that
* feed the critic-calibration projection. A no-op when no [CritiqueFindingsRecordedEvent]s were
* recorded (the LLM-side finding emission is a separate, model-gated activation), and idempotent
* — it skips when outcomes already exist — so it is safe on every terminal path.
*/
internal suspend fun SessionOrchestrator.correlateCritiqueOutcomes(sessionId: SessionId) {
val events = eventStore.read(sessionId)
if (events.any { it.payload is CritiqueOutcomeCorrelatedEvent }) return
val reviews = events.mapNotNull { it.payload as? CritiqueFindingsRecordedEvent }
if (reviews.isEmpty()) return
CritiqueOutcomeCorrelator.correlate(sessionId, reviews).forEach { emit(sessionId, it) }
}
internal suspend fun SessionOrchestrator.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,
),
)
}
/** Appends [payloads] as one transaction (one flow-publish pass) for events that belong together. */
internal suspend fun SessionOrchestrator.emitAll(sessionId: SessionId, payloads: List<EventPayload>) {
if (payloads.size <= 1) {
payloads.firstOrNull()?.let { emit(sessionId, it) }
return
}
eventStore.appendAll(
payloads.map { payload ->
NewEvent(
metadata = EventMetadata(
eventId = EventId(UUID.randomUUID().toString()),
sessionId = sessionId,
timestamp = Clock.System.now(),
schemaVersion = 1,
causationId = null,
correlationId = null,
),
payload = payload,
)
},
)
}
@@ -0,0 +1,159 @@
package com.correx.core.kernel.orchestration
import com.correx.core.events.events.ToolCallAssessedEvent
import com.correx.core.events.events.RepoMapComputedEvent
import com.correx.core.events.events.OutsidePathAccessGrantedEvent
import com.correx.core.events.events.ToolExecutionCompletedEvent
import com.correx.core.events.events.ToolInvocationRequestedEvent
import com.correx.core.events.risk.RiskAction
import com.correx.core.toolintent.SessionContextProjection
import com.correx.core.events.events.TransitionExecutedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.sessions.projections.EgressAllowlistProjection
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.kernel.execution.WorkflowResult
import com.correx.core.tools.contract.ToolCapability
import com.correx.core.transitions.graph.StageConfig
import java.nio.file.Path
import java.nio.file.Paths
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
internal fun SessionOrchestrator.resolveSessionEgressHosts(sessionId: SessionId): Set<String> {
val projection = EgressAllowlistProjection(sessionId)
return eventStore.read(sessionId)
.fold(projection.initial()) { acc, event -> projection.apply(acc, event) }
}
/**
* Folds this session's tool events through [SessionContextProjection] into the read-model the
* anti-hallucination gates consume (reads, writes, active task). Replay-safe (reads the log,
* never live state); empty before anything has happened.
*/
internal fun SessionOrchestrator.resolveSessionContext(sessionId: SessionId): com.correx.core.toolintent.SessionContext {
val projection = SessionContextProjection(sessionId)
return eventStore.read(sessionId)
.fold(projection.initial()) { acc, event -> projection.apply(acc, event) }
}
/**
* The out-of-workspace path a read tool targets, or null if the read stays inside the workspace
* (or there's no workspace policy / path arg). Resolution is purely lexical (absolute-normalize,
* or resolve-against-root for relatives) so the containment decision is deterministic and never
* touches the filesystem. Symlink escapes and privileged locations are caught upstream by the
* plane-2 PathContainmentRule (BLOCK before any prompt), and by the tool's own realize() jail.
*/
internal fun SessionOrchestrator.outsideWorkspaceTarget(parameters: Map<String, Any>, workspaceRoot: Path?): Path? {
workspaceRoot ?: return null
val raw = parameters["path"] as? String ?: return null
val candidate = runCatching { Path.of(raw) }.getOrNull() ?: return null
val root = workspaceRoot.toAbsolutePath().normalize()
val resolved = (if (candidate.isAbsolute) candidate else root.resolve(candidate)).normalize()
return resolved.takeUnless { it.startsWith(root) }
}
/**
* Paths outside the workspace the operator has approved this session (folded from
* [OutsidePathAccessGrantedEvent]). Replay-safe: reads the log, never re-prompts or re-probes.
*/
internal fun SessionOrchestrator.grantedOutsidePaths(sessionId: SessionId): Set<String> =
eventStore.read(sessionId)
.mapNotNull { it.payload as? OutsidePathAccessGrantedEvent }
.filter { it.sessionId == sessionId }
.map { it.path }
.toSet()
/**
* Returns true when the agent must be offered only read-only tools on the next inference turn.
* Active from the moment a READ_BEFORE_WRITE block lands until a FILE_READ completion follows it.
* Derived purely from the event log — no separate flag stored.
*/
internal fun SessionOrchestrator.isReadOnlyMode(sessionId: SessionId): Boolean {
var blocked = false
val pendingReadIds = mutableSetOf<String>()
for (event in eventStore.read(sessionId)) {
when (val p = event.payload) {
is ToolInvocationRequestedEvent ->
if (p.sessionId == sessionId && ToolCapability.FILE_READ in p.capabilities)
pendingReadIds += p.invocationId.value
is ToolCallAssessedEvent ->
if (p.sessionId == sessionId && p.disposition == RiskAction.BLOCK) {
when {
p.issues.any { it.code == READ_BEFORE_WRITE_CODE } -> {
blocked = true
pendingReadIds.clear()
}
// A read BLOCKED because its target doesn't exist (REFERENCE_EXISTS) can never
// produce a completion, so it can't be what lifts read-only mode. Treat the
// attempt as satisfying read-before-write: the file is absent, so there is
// nothing to read before creating it. Without this, a stage that writes a NEW
// file deadlocks — writes are filtered out and every read of the not-yet-created
// target is blocked, burning MAX_TOOL_ROUNDS with no artifact.
p.issues.any { it.code == REFERENCE_EXISTS_CODE } -> {
blocked = false
pendingReadIds.clear()
}
}
}
is ToolExecutionCompletedEvent ->
if (p.sessionId == sessionId && p.invocationId.value in pendingReadIds) {
blocked = false
pendingReadIds.clear()
}
else -> {}
}
}
return blocked
}
internal fun SessionOrchestrator.mandateSuppressedByTicket(sessionId: SessionId, stageId: StageId, stageConfig: StageConfig): Boolean {
if (stageConfig.metadata["role"] == "recovery") return false
return eventStore.read(sessionId)
.mapNotNull { it.payload as? TransitionExecutedEvent }
.lastOrNull { it.to == stageId }
?.transitionId == TICKET_ROUTE
}
/**
* Best real path in the repo map that shares [guessed]'s filename, ranked by shared trailing
* path segments — so a wrong package prefix (e.g. `com/correx/server/...` for the real
* `com/correx/apps/server/...`) still resolves. Null when nothing shares the filename, so we
* only ever suggest a concrete, existing file. Reads the recorded map (replay-safe).
*/
internal fun SessionOrchestrator.suggestClosestPath(sessionId: SessionId, guessed: String): String? {
val map = eventStore.read(sessionId)
.mapNotNull { it.payload as? RepoMapComputedEvent }
.lastOrNull() ?: return null
return closestPathByFilename(guessed, map.entries.map { it.path })
}
/**
* Resolves the actual prompt text a stage will see, for use as the repo-knowledge query.
* `metadata["prompt"]` holds a *path* to the prompt file (resolved into context separately,
* around line 431) — using that path string itself as the embedding query means the query the
* retriever compares against is a filesystem path, not the stage's real instructions, so
* similarity scores never reflect genuine relevance (2026-07-08 finding: no file scored a
* clear match, everything clustered in the embedder's baseline noise band for short strings).
*/
internal suspend fun SessionOrchestrator.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 SessionOrchestrator.isCancelled(sessionId: SessionId): Boolean =
cancellations[sessionId]?.get() == true