refactor(kernel): extract executeStage to extension, promote members internal
Pilot for the SessionOrchestrator god-class decomposition. Moves executeStage (426-line stage-execution driver) out to SessionOrchestratorExecution.kt as an internal extension fun, hoists three nested holders (RunEffectives, ArtifactLadderOutcome, RenderedToolResult) to top-level in SessionOrchestratorTypes.kt, and promotes class/protected members to internal so extensions can reach them. Behavior-preserving relocation; kernel+test+server compile green.
This commit is contained in:
+1
-1
@@ -99,7 +99,7 @@ class DefaultSessionOrchestrator(
|
||||
private val retryCoordinator: RetryCoordinator,
|
||||
artifactStore: ArtifactStore,
|
||||
tokenizer: Tokenizer? = null,
|
||||
private val decisionJournalRepository: DefaultDecisionJournalRepository,
|
||||
decisionJournalRepository: DefaultDecisionJournalRepository,
|
||||
private val compactionService: JournalCompactionService? = null,
|
||||
artifactKindRegistry: ArtifactKindRegistry? = null,
|
||||
repoKnowledgeRetriever: RepoKnowledgeRetriever? = null,
|
||||
|
||||
+127
-704
File diff suppressed because it is too large
Load Diff
+741
@@ -0,0 +1,741 @@
|
||||
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.*
|
||||
import kotlin.coroutines.cancellation.CancellationException
|
||||
|
||||
@Suppress("CyclomaticComplexMethod")
|
||||
internal suspend fun SessionOrchestrator.executeStage(
|
||||
sessionId: SessionId,
|
||||
stageId: StageId,
|
||||
graph: WorkflowGraph,
|
||||
session: Session,
|
||||
config: OrchestrationConfig,
|
||||
effectives: RunEffectives,
|
||||
): StageExecutionResult {
|
||||
val stageConfig = requireNotNull(graph.stages[stageId]) {
|
||||
"Stage '${stageId.value}' not declared in workflow graph"
|
||||
}
|
||||
val systemPrompt = stageConfig.metadata["systemPrompt"]
|
||||
?.let { path ->
|
||||
runCatching { promptResolver.resolve(path) }
|
||||
.onFailure {
|
||||
log.error(error(stageId, path, it))
|
||||
}
|
||||
.getOrNull()
|
||||
}
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?.let { text ->
|
||||
listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L0,
|
||||
content = text,
|
||||
sourceType = "systemPrompt",
|
||||
sourceId = stageId.value,
|
||||
tokenEstimate = estimateTokens(text),
|
||||
role = EntryRole.SYSTEM,
|
||||
),
|
||||
)
|
||||
} ?: config.defaultSystemPromptPath
|
||||
?.let { path ->
|
||||
runCatching { promptResolver.resolve(path) }
|
||||
.onFailure {
|
||||
log.error(error(stageId, path, it))
|
||||
}
|
||||
.getOrNull()
|
||||
}
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?.let { text ->
|
||||
listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L0,
|
||||
content = text,
|
||||
sourceType = "systemPrompt",
|
||||
sourceId = stageId.value,
|
||||
tokenEstimate = estimateTokens(text),
|
||||
role = EntryRole.SYSTEM,
|
||||
),
|
||||
)
|
||||
} ?: emptyList()
|
||||
// A stage re-entered to repair a gate failure has its own (often generative/"scaffold") prompt
|
||||
// SUPPRESSED: that mandate is what drove it to overwrite real files with stubs on re-entry. The
|
||||
// recovery-ticket entry (buildRecoveryTicketEntry) is the sole mandate here — it already carries
|
||||
// "make the smallest change, don't rewrite from scratch" plus the gate evidence. See
|
||||
// mandateSuppressedByTicket.
|
||||
val promptEntries = if (mandateSuppressedByTicket(sessionId, stageId, stageConfig)) {
|
||||
emptyList()
|
||||
} else {
|
||||
stageConfig.metadata["promptInline"]
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?.let { text ->
|
||||
listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L1,
|
||||
content = text,
|
||||
sourceType = "agentPrompt",
|
||||
sourceId = stageId.value,
|
||||
tokenEstimate = estimateTokens(text),
|
||||
role = EntryRole.USER,
|
||||
),
|
||||
)
|
||||
}
|
||||
?: stageConfig.metadata["prompt"]
|
||||
?.let { path ->
|
||||
val resolvedText = runCatching { promptResolver.resolve(path) }
|
||||
.onFailure {
|
||||
log.error(
|
||||
"[SessionOrchestrator] stage=${stageId.value}: " +
|
||||
"failed to load prompt '$path': ${it.message}",
|
||||
)
|
||||
}
|
||||
.getOrNull()
|
||||
val text = resolvedText?.takeIf { it.isNotBlank() }
|
||||
?: error(
|
||||
"[SessionOrchestrator] stage=${stageId.value}: " +
|
||||
"declared prompt '$path' could not be resolved",
|
||||
)
|
||||
listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L1,
|
||||
content = text,
|
||||
sourceType = "agentPrompt",
|
||||
sourceId = stageId.value,
|
||||
tokenEstimate = estimateTokens(text),
|
||||
role = EntryRole.USER,
|
||||
),
|
||||
)
|
||||
}
|
||||
?: emptyList()
|
||||
}
|
||||
|
||||
val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted }
|
||||
require(llmEmittedSlots.size <= 1) {
|
||||
"Stage '${stageId.value}' declares ${llmEmittedSlots.size} LLM-emitted artifact kinds; " +
|
||||
"only 0 or 1 is supported"
|
||||
}
|
||||
val responseFormat = llmEmittedSlots.firstOrNull()
|
||||
?.let { ResponseFormat.Json(it.kind.deriveJsonSchema()) }
|
||||
?: ResponseFormat.Text
|
||||
|
||||
val schemaEntries = buildSchemaEntries(responseFormat, stageId)
|
||||
val steeringEntries = buildSteeringNoteEntries(sessionId)
|
||||
val clarificationEntries = buildClarificationAnswerEntries(sessionId)
|
||||
|
||||
// Maintain the running transcript in true chronological order. Reading it back
|
||||
// from currentContext.layers (a Map grouped by layer) would scramble turn order
|
||||
// across rounds; instead we grow our own ordered list and let the builder restamp
|
||||
// ordinals from it each round.
|
||||
val journalState = decisionJournalRepository.getJournal(sessionId)
|
||||
val summaryText = journalState.summaryArtifactId?.let { id ->
|
||||
artifactStore.get(id)?.toString(Charsets.UTF_8)
|
||||
?: run {
|
||||
log.warn("Journal summary artifact {} not found in store — rendering without summary", id.value)
|
||||
null
|
||||
}
|
||||
}
|
||||
val journalText = decisionJournalRenderer.render(journalState, summaryText)
|
||||
val journalEntries = if (journalText.isBlank()) emptyList() else listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L0,
|
||||
content = journalText,
|
||||
sourceType = "decisionJournal",
|
||||
sourceId = "decision-journal",
|
||||
tokenEstimate = estimateTokens(journalText),
|
||||
role = EntryRole.SYSTEM,
|
||||
),
|
||||
)
|
||||
val repoMapEntries = buildContextualRepoEntries(sessionId, stageId, stageConfig)
|
||||
val sessionEvents = eventStore.read(sessionId)
|
||||
val criticIds = criticArtifactIds(sessionEvents, graph, stageId)
|
||||
val criticFrom = sessionEvents
|
||||
.mapNotNull { it.payload as? RefinementIterationEvent }
|
||||
.lastOrNull { it.cycleKey.endsWith("->${stageId.value}") }
|
||||
?.cycleKey?.substringBefore("->")
|
||||
val needsEntries = buildNeedsArtifactEntries(sessionId, stageConfig, criticIds, criticFrom)
|
||||
val profileEntries = session.state.boundProfile?.about
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?.let { about ->
|
||||
listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L0,
|
||||
content = "## Operator profile\n$about",
|
||||
sourceType = "operatorProfile",
|
||||
sourceId = "operator-profile",
|
||||
tokenEstimate = about.length / 4,
|
||||
role = EntryRole.SYSTEM,
|
||||
),
|
||||
)
|
||||
} ?: emptyList()
|
||||
val projectProfileEntries = session.state.boundProjectProfile
|
||||
?.let { listOf(buildProjectProfileEntry(it)) } ?: emptyList()
|
||||
// AGENTS.md injection is disabled for now. The workspace-root AGENTS.md (the DOX framework)
|
||||
// carries a mandatory "Read Before Editing" walk — "read the whole AGENTS.md chain, do not
|
||||
// rely on memory, re-read before editing" — which drives a small model into an endless
|
||||
// read-loop that never writes (2026-07-05 root cause). Reincorporating DOX properly (scoped
|
||||
// to the stage's subproject, not the repo root; not read-mandating) is deferred.
|
||||
val agentInstructionsEntries = emptyList<ContextEntry>()
|
||||
val retryFeedbackEntries = buildRetryFeedbackEntry(sessionEvents, stageId)
|
||||
?.let { listOf(it) } ?: emptyList()
|
||||
val recoveryTicketEntries = buildRecoveryTicketEntry(sessionEvents, stageId)
|
||||
?.let { listOf(it) } ?: emptyList()
|
||||
val vocabularyEntries = artifactKindRegistry
|
||||
?.takeIf { stageConfig.metadata["injectArtifactKinds"] == "true" }
|
||||
?.let { listOf(buildArtifactKindVocabularyEntry(it.list())) } ?: emptyList()
|
||||
// Deterministic claim: a stage flagged claimTask gets the kernel to claim the next ready task
|
||||
// (or re-surface the one already claimed) and inject its context bundle as L0 — the loop
|
||||
// advances by recorded fact, not by the agent remembering to claim.
|
||||
val claimedTaskEntries = if (stageConfig.metadata["claimTask"] == "true") {
|
||||
taskClaimCoordinator?.claimNext(sessionId)?.let { bundle ->
|
||||
listOf(
|
||||
ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L0,
|
||||
content = bundle,
|
||||
sourceType = "claimedTask",
|
||||
sourceId = stageId.value,
|
||||
tokenEstimate = estimateTokens(bundle),
|
||||
role = EntryRole.SYSTEM,
|
||||
),
|
||||
)
|
||||
} ?: emptyList()
|
||||
} else {
|
||||
emptyList()
|
||||
}
|
||||
// Remaining-delta checklist (stage-termination design 2026-07-11 §1a): the stage's currently
|
||||
// unmet file-contract assertions as a forward-looking, shrinking TODO. Injected pinned
|
||||
// (neverDrop) so it survives the budget/dedup passes that otherwise wipe the model's memory of
|
||||
// progress; refreshed inside the tool loop each time a write lands (§1b, cache-until-write).
|
||||
var remainingDeltaResults = evaluateStageContract(sessionId, stageId, stageConfig, effectives)
|
||||
val remainingDeltaEntries = remainingDeltaResults
|
||||
?.let { buildRemainingDeltaEntry(contractFailureItems(it)) }
|
||||
?.let { listOf(it) } ?: emptyList()
|
||||
var accumulatedEntries = stampBuckets(
|
||||
systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries +
|
||||
journalEntries + repoMapEntries + claimedTaskEntries +
|
||||
needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries +
|
||||
clarificationEntries + retryFeedbackEntries + recoveryTicketEntries + remainingDeltaEntries,
|
||||
)
|
||||
val contextPack = runCatching {
|
||||
contextPackBuilder.build(
|
||||
id = ContextPackId(UUID.randomUUID().toString()),
|
||||
sessionId = sessionId,
|
||||
stageId = stageId,
|
||||
entries = accumulatedEntries,
|
||||
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
||||
)
|
||||
}.getOrElse { e ->
|
||||
if (e is CancellationException) throw e
|
||||
if (e !is RequiredContextOverflowException) throw e
|
||||
// Required context doesn't fit — retrying with the same inputs can't succeed.
|
||||
log.error("[Orchestrator] stage={}: {}", stageId.value, e.message)
|
||||
return StageExecutionResult.Failure(e.message ?: "required context overflow", retryable = false)
|
||||
}
|
||||
emitContextTruncationIfNeeded(sessionId, stageId, contextPack)
|
||||
|
||||
var currentContext = contextPack
|
||||
var inferenceResult = runInference(
|
||||
sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
|
||||
)
|
||||
var toolRounds = 0
|
||||
var consecutiveReadOnlyRounds = 0
|
||||
var consecutiveRejectedRounds = 0
|
||||
// Fingerprints ("tool:arguments") of every read-only call made during the current
|
||||
// no-write streak. A round only counts against the read-loop breaker if it repeats calls
|
||||
// already seen — a stage reading N distinct files before writing is legitimate context
|
||||
// gathering, not a loop, and shouldn't trip the same counter as re-reading the same file.
|
||||
val seenReadFingerprints = mutableSetOf<String>()
|
||||
// Set when the model produces its artifact via the emit_artifact tool instead of a final
|
||||
// JSON message; overrides the post-loop capture of the (then-empty) assistant text.
|
||||
var llmArtifactOverride: String? = null
|
||||
|
||||
val fileWrittenSlots = stageConfig.produces.filter { it.kind.id == "file_written" }
|
||||
fun owesFileWrite(): Boolean = fileWrittenSlots.isNotEmpty() &&
|
||||
fileWrittenSlots.any { artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank() }
|
||||
|
||||
// Gate: if the stage declares require_task_decompose, block stage_complete until at least one
|
||||
// task_decompose (or task_create) call has completed successfully in this stage. Checked via
|
||||
// ToolInvocationRequestedEvent + ToolExecutionCompletedEvent in the session event log.
|
||||
// ponytail: scans events on every stage_complete attempt; negligible cost (in-memory list, rare call).
|
||||
fun owesTaskDecompose(): Boolean {
|
||||
if (stageConfig.metadata["requireTaskDecompose"] != "true") return false
|
||||
val events = eventStore.read(sessionId)
|
||||
val requestedIds = events
|
||||
.mapNotNull { it.payload as? ToolInvocationRequestedEvent }
|
||||
.filter { it.stageId == stageId && it.toolName in setOf("task_decompose", "task_create") }
|
||||
.map { it.invocationId }
|
||||
.toSet()
|
||||
if (requestedIds.isEmpty()) return true
|
||||
val completedIds = events
|
||||
.mapNotNull { it.payload as? ToolExecutionCompletedEvent }
|
||||
.filter { it.toolName in setOf("task_decompose", "task_create") && it.invocationId in requestedIds }
|
||||
.map { it.invocationId }
|
||||
.toSet()
|
||||
return completedIds.isEmpty()
|
||||
}
|
||||
|
||||
// Append a corrective tool-result and re-run inference (bounded by MAX_TOOL_ROUNDS).
|
||||
// Returns the new result rather than mutating inferenceResult, to preserve smart casts.
|
||||
suspend fun pushBack(nudge: String, forceWriteOnly: Boolean = false): InferenceResult {
|
||||
accumulatedEntries = accumulatedEntries + ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L2,
|
||||
sourceType = "toolResult",
|
||||
sourceId = UUID.randomUUID().toString(),
|
||||
content = nudge,
|
||||
tokenEstimate = estimateTokens(nudge),
|
||||
role = EntryRole.TOOL,
|
||||
)
|
||||
currentContext = contextPackBuilder.build(
|
||||
id = ContextPackId(UUID.randomUUID().toString()),
|
||||
sessionId = sessionId,
|
||||
stageId = stageId,
|
||||
entries = accumulatedEntries,
|
||||
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
||||
)
|
||||
emitContextTruncationIfNeeded(sessionId, stageId, currentContext)
|
||||
toolRounds++
|
||||
return runInference(
|
||||
sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
|
||||
forceWriteOnly = forceWriteOnly,
|
||||
)
|
||||
}
|
||||
|
||||
val writeNudge = "ERROR: you described the change but did not apply it. You MUST call the " +
|
||||
"file_edit or file_write tool to write the change to the file before finishing — do " +
|
||||
"not output the file contents as a message."
|
||||
|
||||
val readLoopNudge = "STOP reading. You have read enough files and this stage has produced " +
|
||||
"nothing yet. You MUST now call the file_write tool to create the required file(s) for " +
|
||||
"this stage. Do not call file_read or list_dir again before you have written a file."
|
||||
|
||||
while (
|
||||
inferenceResult is InferenceResult.Success &&
|
||||
toolRounds < tuning.maxToolRounds &&
|
||||
(inferenceResult.response.finishReason is FinishReason.ToolCall || owesFileWrite())
|
||||
) {
|
||||
// Content turn (no tool call) but the stage still owes a file_written artifact: the
|
||||
// model emitted the change as prose instead of invoking the write tool. Nudge it to
|
||||
// actually write (F-018).
|
||||
if (inferenceResult.response.finishReason !is FinishReason.ToolCall) {
|
||||
inferenceResult = pushBack(writeNudge)
|
||||
continue
|
||||
}
|
||||
// emit_artifact is a meta-tool: the LLM calls it with the artifact's fields as arguments.
|
||||
// Capture those arguments as the artifact content and exit the loop straight to validation
|
||||
// (the post-loop capture honours this override instead of the empty assistant text).
|
||||
val emitCall = inferenceResult.response.toolCalls.firstOrNull { it.function.name == EMIT_ARTIFACT_TOOL }
|
||||
if (emitCall != null && llmEmittedSlots.isNotEmpty()) {
|
||||
llmArtifactOverride = emitCall.function.arguments
|
||||
break
|
||||
}
|
||||
// stage_complete is a meta-tool: the LLM calls it to signal the stage's goal is met.
|
||||
// Skip tool dispatch — the loop exits and normal validation/transition follows
|
||||
// (emitToolArtifacts + verifyProduces run on the success path after the loop).
|
||||
if (inferenceResult.response.toolCalls.any { it.function.name == STAGE_COMPLETE_TOOL }) {
|
||||
// Premature-completion guard: don't accept stage_complete while the stage still
|
||||
// owes an artifact. For an LLM-emitted slot, nudge to emit the JSON (F-010); for a
|
||||
// file_written slot, nudge to invoke the write tool (F-018). Otherwise complete.
|
||||
val owedLlm = llmEmittedSlots.firstOrNull()?.takeIf {
|
||||
inferenceResult.response.text.isBlank() &&
|
||||
artifactContentCache["${sessionId.value}:${it.name.value}"].isNullOrBlank()
|
||||
}
|
||||
when {
|
||||
owedLlm != null -> inferenceResult = pushBack(
|
||||
"ERROR: stage_complete is not allowed yet — you have not emitted the " +
|
||||
"required '${owedLlm.name.value}' artifact. Call the emit_artifact tool with " +
|
||||
"the fields filled in (or respond with the artifact as a single JSON message " +
|
||||
"matching the provided schema).",
|
||||
)
|
||||
owesFileWrite() -> inferenceResult = pushBack(writeNudge)
|
||||
owesTaskDecompose() -> inferenceResult = pushBack(
|
||||
"ERROR: stage_complete is not allowed yet — this stage requires a successful " +
|
||||
"task_decompose (or task_create) call before it can complete. Call task_decompose " +
|
||||
"with a parent epic and DEPENDS_ON-linked children representing the full task graph. " +
|
||||
"Do not call stage_complete until task_decompose has returned successfully.",
|
||||
)
|
||||
else -> break
|
||||
}
|
||||
continue
|
||||
}
|
||||
val toolEntries = dispatchToolCalls(sessionId, stageId,
|
||||
inferenceResult.response.toolCalls, stageConfig, effectives,
|
||||
approvalModeFor(session.state.boundProfile?.approvalMode),
|
||||
inferenceResult.response.reasoning)
|
||||
val fatalEntry = toolEntries.firstOrNull { it.content.startsWith("FATAL:") }
|
||||
if (fatalEntry != null) {
|
||||
emitProcessResultEvents(sessionId, stageId, stageConfig)
|
||||
return StageExecutionResult.Failure(
|
||||
fatalEntry.content.removePrefix("FATAL: "),
|
||||
retryable = false,
|
||||
)
|
||||
}
|
||||
// Recoverable (ERROR:) tool failures are NOT terminal — they flow back into the
|
||||
// loop as tool-result context so the model can see the error and adapt (bounded by
|
||||
// MAX_TOOL_ROUNDS). Only FATAL: failures (handled above) abort the stage.
|
||||
accumulatedEntries = accumulatedEntries + toolEntries
|
||||
// Read-loop breaker: this round called only read-only tools yet the stage still owes a
|
||||
// file_written artifact. Left alone the model keeps reading until MAX_TOOL_ROUNDS and
|
||||
// never writes (F-018 nudges only cover a prose turn or a premature stage_complete, not
|
||||
// an endless read-loop). After READ_LOOP_NUDGE_THRESHOLD such rounds, force the write
|
||||
// nudge; a real write resets the counter so multi-file stages are unaffected.
|
||||
if (owesFileWrite() &&
|
||||
inferenceResult.response.toolCalls.none { it.function.name in WRITE_TOOL_NAMES }
|
||||
) {
|
||||
val roundFingerprints = inferenceResult.response.toolCalls
|
||||
.map { "${it.function.name}:${it.function.arguments}" }
|
||||
val madeProgress = roundFingerprints.any { it !in seenReadFingerprints }
|
||||
seenReadFingerprints += roundFingerprints
|
||||
if (madeProgress) {
|
||||
consecutiveReadOnlyRounds = 0
|
||||
} else {
|
||||
consecutiveReadOnlyRounds++
|
||||
if (consecutiveReadOnlyRounds >= tuning.readLoopNudgeThreshold) {
|
||||
consecutiveReadOnlyRounds = 0
|
||||
inferenceResult = pushBack(readLoopNudge, forceWriteOnly = true)
|
||||
continue
|
||||
}
|
||||
}
|
||||
} else {
|
||||
consecutiveReadOnlyRounds = 0
|
||||
seenReadFingerprints.clear()
|
||||
}
|
||||
// Rejection-loop breaker (stage-type agnostic): every tool result this round was a
|
||||
// rejection (plane-2 BLOCKED or a stage/policy ERROR) and nothing succeeded. A stage that
|
||||
// owes no file_written artifact (discovery, analyst) never trips the read-loop breaker, so
|
||||
// without this it re-issues the same rejected call until MAX_TOOL_ROUNDS. After the
|
||||
// threshold, tell it to stop retrying and produce its output; any success resets.
|
||||
val toolResults = toolEntries.filter { it.role == EntryRole.TOOL }
|
||||
val allRejected = toolResults.isNotEmpty() &&
|
||||
toolResults.all { it.content.startsWith("BLOCKED:") || it.content.startsWith("ERROR:") }
|
||||
if (allRejected) {
|
||||
consecutiveRejectedRounds++
|
||||
if (consecutiveRejectedRounds >= tuning.rejectionLoopNudgeThreshold) {
|
||||
consecutiveRejectedRounds = 0
|
||||
inferenceResult = pushBack(
|
||||
"STOP. Your last tool calls were all rejected and retrying the same paths will " +
|
||||
"keep failing. Do not repeat them. Proceed with the information you already " +
|
||||
"have: produce this stage's required output now (call stage_complete, or emit " +
|
||||
"the required artifact). If a path you wanted does not exist yet, that is a " +
|
||||
"fact to record — do not keep probing for it.",
|
||||
)
|
||||
continue
|
||||
}
|
||||
} else {
|
||||
consecutiveRejectedRounds = 0
|
||||
}
|
||||
// Refresh the remaining-delta checklist only when a write actually landed this round
|
||||
// (cache-until-write, §1b): reads can never change the contract verdict, so re-evaluating
|
||||
// on them would just burn filesystem work and re-emit identical events. On a successful
|
||||
// write, recompute contract-minus-reality and swap the pinned entry so the model watches
|
||||
// the list shrink toward the empty=done signal it previously lacked.
|
||||
val wroteThisRound = inferenceResult.response.toolCalls.any { it.function.name in WRITE_TOOL_NAMES } &&
|
||||
toolEntries.any {
|
||||
it.role == EntryRole.TOOL &&
|
||||
!it.content.startsWith("ERROR:") && !it.content.startsWith("BLOCKED:")
|
||||
}
|
||||
if (wroteThisRound) {
|
||||
remainingDeltaResults = evaluateStageContract(sessionId, stageId, stageConfig, effectives)
|
||||
val refreshed = remainingDeltaResults?.let { buildRemainingDeltaEntry(contractFailureItems(it)) }
|
||||
accumulatedEntries = accumulatedEntries.filterNot { it.sourceType == "remainingDelta" } +
|
||||
listOfNotNull(refreshed)
|
||||
}
|
||||
currentContext = contextPackBuilder.build(
|
||||
id = ContextPackId(UUID.randomUUID().toString()),
|
||||
sessionId = sessionId,
|
||||
stageId = stageId,
|
||||
entries = accumulatedEntries,
|
||||
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
||||
)
|
||||
emitContextTruncationIfNeeded(sessionId, stageId, currentContext)
|
||||
inferenceResult = runInference(
|
||||
sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, effectives,
|
||||
)
|
||||
toolRounds++
|
||||
}
|
||||
// Final clean emission. Producing the artifact while tools are in the request is unreliable on
|
||||
// some local chat templates (Gemma leaks `<|channel>` markers into the text; other models keep
|
||||
// tool-calling until the round cap and never emit). For an LLM-emitted stage that hasn't already
|
||||
// produced its artifact via the emit_artifact tool, fire ONE tools-less inference — verified to
|
||||
// yield clean JSON deterministically. Skipped when the model already emitted clean final text.
|
||||
if (inferenceResult is InferenceResult.Success && llmEmittedSlots.isNotEmpty() && llmArtifactOverride == null) {
|
||||
val last = inferenceResult.response
|
||||
val needsCleanEmission = last.finishReason is FinishReason.ToolCall ||
|
||||
last.text.isBlank() || last.text.contains("<|channel")
|
||||
if (needsCleanEmission && !isCancelled(sessionId)) {
|
||||
val nudge = "Stop calling tools. Output the required '${llmEmittedSlots.first().name.value}' " +
|
||||
"artifact now as a single JSON object matching the schema — no tool calls, no commentary."
|
||||
accumulatedEntries = accumulatedEntries + ContextEntry(
|
||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||
layer = ContextLayer.L2,
|
||||
sourceType = "toolResult",
|
||||
sourceId = UUID.randomUUID().toString(),
|
||||
content = nudge,
|
||||
tokenEstimate = estimateTokens(nudge),
|
||||
role = EntryRole.TOOL,
|
||||
)
|
||||
currentContext = contextPackBuilder.build(
|
||||
id = ContextPackId(UUID.randomUUID().toString()),
|
||||
sessionId = sessionId,
|
||||
stageId = stageId,
|
||||
entries = accumulatedEntries,
|
||||
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
||||
)
|
||||
inferenceResult = runInference(
|
||||
sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs,
|
||||
responseFormat, effectives, withTools = false,
|
||||
)
|
||||
}
|
||||
}
|
||||
return when (inferenceResult) {
|
||||
is InferenceResult.Cancelled -> StageExecutionResult.Failure("CANCELLED", retryable = false)
|
||||
is InferenceResult.Failed ->
|
||||
StageExecutionResult.Failure(inferenceResult.reason, retryable = isRetryableFailure(inferenceResult.reason))
|
||||
is InferenceResult.Success -> {
|
||||
if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false)
|
||||
// Capture the LLM-emitted artifact's JSON so transition conditions
|
||||
// (e.g. ArtifactFieldEquals on a reviewer's verdict) can read it. An emit_artifact
|
||||
// tool call (llmArtifactOverride) supplies the content directly; otherwise the
|
||||
// artifact is the final assistant message text.
|
||||
val rawArtifactText = llmArtifactOverride ?: inferenceResult.response.text
|
||||
val slot = llmEmittedSlots.firstOrNull()
|
||||
// Artifact-emission ladder: deterministic repair (prose/fences/trailing commas), then
|
||||
// — rarely, gated by the failure classifier — one grammar-constrained LLM-repair rung,
|
||||
// before the text is cached/validated. On a hard failure the policy decision
|
||||
// short-circuits the stage. Only the nondeterministic rung records events (spec 2026-07-04 §6).
|
||||
val artifactText = if (slot != null) {
|
||||
when (val res = artifactExtractionPipeline.run(rawArtifactText, slot.kind.deriveJsonSchema())) {
|
||||
is ArtifactExtractionPipeline.ExtractionResult.Resolved -> {
|
||||
if (res.repaired) {
|
||||
emitArtifactRepairAttempted(sessionId, stageId, slot, ArtifactFailure.FORMATTING, "DETERMINISTIC")
|
||||
emitArtifactRepairResolved(sessionId, stageId, slot, res.canonicalJson.toString())
|
||||
}
|
||||
res.canonicalJson.toString()
|
||||
}
|
||||
is ArtifactExtractionPipeline.ExtractionResult.Unresolved ->
|
||||
when (val ladder = repairArtifact(sessionId, stageId, slot, res, stageConfig, effectives, config.stageTimeoutMs)) {
|
||||
is ArtifactLadderOutcome.Text -> ladder.text
|
||||
is ArtifactLadderOutcome.Reject -> return ladder.failure
|
||||
}
|
||||
}
|
||||
} else {
|
||||
rawArtifactText
|
||||
}
|
||||
val artifactHash = if (artifactText != rawArtifactText || llmArtifactOverride != null) {
|
||||
artifactStore.put(artifactText.toByteArray())
|
||||
} else {
|
||||
inferenceResult.responseArtifactId
|
||||
}
|
||||
slot?.let { slot ->
|
||||
artifactContentCache["${sessionId.value}:${slot.name.value}"] = artifactText
|
||||
artifactHash?.let { hash ->
|
||||
emit(
|
||||
sessionId,
|
||||
ArtifactContentStoredEvent(
|
||||
artifactId = slot.name,
|
||||
contentHash = hash,
|
||||
sessionId = sessionId,
|
||||
stageId = stageId,
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
val validationCtx = ValidationContext(
|
||||
graph = graph,
|
||||
sessionState = session.state,
|
||||
availableTools = effectives.registry?.all()?.map { it.name }?.toSet(),
|
||||
producedArtifactContent = graph.stages.values.flatMap { it.produces }.mapNotNull { slot ->
|
||||
artifactContentCache["${sessionId.value}:${slot.name.value}"]?.let { slot.name.value to it }
|
||||
}.toMap(),
|
||||
)
|
||||
when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) {
|
||||
is StageExecutionResult.Success -> {
|
||||
emitToolArtifacts(sessionId, stageId, stageConfig)
|
||||
emitLlmArtifacts(sessionId, stageId, stageConfig)
|
||||
runPostStageGates(
|
||||
sessionId, stageId, stageConfig, effectives,
|
||||
session.state.boundProjectProfile?.commands ?: emptyMap(),
|
||||
)
|
||||
}
|
||||
|
||||
is StageExecutionResult.Failure -> outcome
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+22
@@ -0,0 +1,22 @@
|
||||
package com.correx.core.kernel.orchestration
|
||||
|
||||
import com.correx.core.toolintent.WorkspacePolicy
|
||||
import com.correx.core.tools.contract.ToolExecutor
|
||||
import com.correx.core.tools.registry.ToolRegistry
|
||||
import com.correx.core.transitions.execution.StageExecutionResult
|
||||
|
||||
/** Per-run effective tool wiring (workspace-scoped registry/executor/policy or the orchestrator defaults). */
|
||||
internal data class RunEffectives(
|
||||
val registry: ToolRegistry?,
|
||||
val executor: ToolExecutor?,
|
||||
val policy: WorkspacePolicy?,
|
||||
)
|
||||
|
||||
/** Result of the artifact extraction/repair ladder: usable text, or a hard stage failure. */
|
||||
internal sealed interface ArtifactLadderOutcome {
|
||||
data class Text(val text: String) : ArtifactLadderOutcome
|
||||
data class Reject(val failure: StageExecutionResult.Failure) : ArtifactLadderOutcome
|
||||
}
|
||||
|
||||
/** A tool result rendered for the model, with the CAS hash of the full (pre-truncation) output if spilled. */
|
||||
internal data class RenderedToolResult(val content: String, val fullOutputHash: String?)
|
||||
Reference in New Issue
Block a user