diff --git a/core/context/src/main/kotlin/com/correx/core/context/builder/DefaultContextPackBuilder.kt b/core/context/src/main/kotlin/com/correx/core/context/builder/DefaultContextPackBuilder.kt index 99f2d2e6..87dd2b22 100644 --- a/core/context/src/main/kotlin/com/correx/core/context/builder/DefaultContextPackBuilder.kt +++ b/core/context/src/main/kotlin/com/correx/core/context/builder/DefaultContextPackBuilder.kt @@ -15,10 +15,12 @@ import com.correx.core.context.compression.ToMeMerger import com.correx.core.context.compression.TokenPruner import com.correx.core.context.compression.UniformRelevanceScorer import com.correx.core.context.model.CompressionMetadata +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.EntryRole +import com.correx.core.context.model.OptionalCeilings import com.correx.core.context.model.TokenBudget import com.correx.core.events.types.ContextEntryId import com.correx.core.events.types.ContextPackId @@ -43,9 +45,13 @@ class DefaultContextPackBuilder( private val tokenPruner: TokenPruner = NoOpTokenPruner, private val toMeMerger: ToMeMerger = ToMeMerger(), private val relevanceScorer: RelevanceScorer = UniformRelevanceScorer, + private val ceilings: OptionalCeilings = OptionalCeilings(), ) : ContextPackBuilder { - private val neverDropSourceTypes = setOf("steeringNote", "eventHistory", "decisionJournal", "factSheet") + // decisionJournal is deliberately NOT here anymore: the journal is OPTIONAL-bucket, capped + // by OptionalCeilings. Pinning is otherwise bucket-driven (ContextBucket.REQUIRED); this set + // only covers entries built outside the orchestrator that never stamp buckets. + private val neverDropSourceTypes = setOf("steeringNote", "eventHistory", "factSheet") private companion object { const val CHARS_PER_TOKEN = 4 @@ -95,15 +101,34 @@ class DefaultContextPackBuilder( if (facts.isEmpty()) merged else merged + factSheetEntry(factSheet.render(facts)) } else merged - val (pinned, compressible) = withFactSheet.partition { it.sourceType in neverDropSourceTypes } + val (pinned, compressible) = withFactSheet.partition { + it.bucket == ContextBucket.REQUIRED || it.sourceType in neverDropSourceTypes + } + // Required entries are never trimmed: if they alone blow the budget, fail fast rather + // than ship a silently truncated instruction set. + val required = withFactSheet.filter { it.bucket == ContextBucket.REQUIRED } + val requiredTokens = required.sumOf { it.tokenEstimate } + if (requiredTokens > budget.limit) { + throw RequiredContextOverflowException( + requiredTokens = requiredTokens, + budgetLimit = budget.limit, + breakdown = required.groupBy { it.sourceType } + .mapValues { (_, group) -> group.sumOf { it.tokenEstimate } }, + ) + } val pinnedTokens = pinned.sumOf { it.tokenEstimate } var remainingTokens = (budget.limit - pinnedTokens).coerceAtLeast(0) + // Optional per-type ceilings are enforced by truncation here, not left to the group + // compressor: several strategies (EventHistory) pass entries through untouched, and these + // feeds are single large entries — dropping whole entries would lose them entirely. + val ceilinged = compressible.map(::applyCeiling) + // Query-conditioned selection: when a real relevance scorer is wired, reorder each freeform // group so the least-relevant turns sit at the front (dropped first) and the most-relevant // land at the tail the Conversation strategy keeps. Final layering re-sorts by ordinal, so // chronological render order is unaffected — only *what survives* the budget changes. - val relevanceRanked = rankByRelevance(compressible, currentQuery(withFactSheet)) + val relevanceRanked = rankByRelevance(ceilinged, currentQuery(withFactSheet)) // Stage 9 (BUDGET_ALLOCATOR_STRICT): compress structured groups before freeform so a long // conversation can't starve tool logs / artifacts out of the window. @@ -116,7 +141,12 @@ class DefaultContextPackBuilder( val compressed = orderedGroups.flatMap { (sourceType, group) -> val strategy = strategyFor(sourceType) strategiesUsed += strategy::class.simpleName ?: "Unknown" - val result = compressor.compress(group, TokenBudget(limit = remainingTokens.coerceAtLeast(0)), strategy) + // Optional groups run under their per-type ceiling so no single feed (repo map, + // journal, profile) can crowd out the rest of the optional block. + val groupLimit = ceilings.capFor(sourceType) + ?.coerceAtMost(remainingTokens.coerceAtLeast(0)) + ?: remainingTokens.coerceAtLeast(0) + val result = compressor.compress(group, TokenBudget(limit = groupLimit), strategy) remainingTokens -= result.sumOf { it.tokenEstimate } result } @@ -151,6 +181,8 @@ class DefaultContextPackBuilder( .map { it.ordinal }.sortedDescending() val tier0 = if (tierSplit) freeformOrdinals.take(TIER0_TURNS).toSet() else emptySet() return entries.map { entry -> + // Required entries are exempt from lossy pruning by definition of the bucket. + if (entry.bucket == ContextBucket.REQUIRED) return@map entry val isFreeform = classifier.classify(entry) == ContextClass.FREEFORM && entry.ordinal !in tier0 val isLargeDoc = entry.sourceType in DOC_SOURCE_TYPES && entry.content.length >= DOC_MIN_CHARS when { @@ -162,10 +194,30 @@ class DefaultContextPackBuilder( } } + /** + * Truncates an OPTIONAL entry to its per-sourceType token ceiling. The decision journal is + * chronological (newest decisions at the tail), so it keeps its tail; every other feed keeps + * its head (headers/most-structural lines first). + */ + private fun applyCeiling(entry: ContextEntry): ContextEntry { + val cap = ceilings.capFor(entry.sourceType) ?: return entry + if (entry.tokenEstimate <= cap) return entry + val trimmedTailMarker = "…(older entries trimmed)\n" + val trimmedHeadMarker = "\n…(trimmed to budget)" + val content = if (entry.sourceType == "decisionJournal") { + trimmedTailMarker + entry.content.takeLast(cap * CHARS_PER_TOKEN - trimmedTailMarker.length) + } else { + entry.content.take(cap * CHARS_PER_TOKEN - trimmedHeadMarker.length) + trimmedHeadMarker + } + return entry.copy(content = content, tokenEstimate = estimateTokens(content)) + } + private fun spansOf(entry: ContextEntry) = tagger.protectedSpans(entry.content) private fun applyToMe(entries: List): List { - val (freeform, rest) = entries.partition { classifier.classify(it) == ContextClass.FREEFORM } + val (freeform, rest) = entries.partition { + it.bucket != ContextBucket.REQUIRED && classifier.classify(it) == ContextClass.FREEFORM + } return (rest + toMeMerger.merge(freeform)).sortedBy { it.ordinal } } diff --git a/core/context/src/main/kotlin/com/correx/core/context/builder/RequiredContextOverflowException.kt b/core/context/src/main/kotlin/com/correx/core/context/builder/RequiredContextOverflowException.kt new file mode 100644 index 00000000..5d515d9b --- /dev/null +++ b/core/context/src/main/kotlin/com/correx/core/context/builder/RequiredContextOverflowException.kt @@ -0,0 +1,15 @@ +package com.correx.core.context.builder + +/** + * The REQUIRED bucket alone (task, constraints, loaded inputs, failures) does not fit the + * stage's token budget. Required context is never trimmed, so the only correct outcome is + * failing the build loudly — a silently truncated instruction is worse than no run. + */ +class RequiredContextOverflowException( + val requiredTokens: Int, + val budgetLimit: Int, + val breakdown: Map, +) : IllegalStateException( + "Required context ($requiredTokens tokens) exceeds the stage budget ($budgetLimit); " + + "by sourceType: " + breakdown.entries.joinToString(", ") { "${it.key}=${it.value}" }, +) diff --git a/core/context/src/main/kotlin/com/correx/core/context/model/ContextEntry.kt b/core/context/src/main/kotlin/com/correx/core/context/model/ContextEntry.kt index 016aa326..f689868c 100644 --- a/core/context/src/main/kotlin/com/correx/core/context/model/ContextEntry.kt +++ b/core/context/src/main/kotlin/com/correx/core/context/model/ContextEntry.kt @@ -5,6 +5,15 @@ import kotlinx.serialization.Serializable enum class EntryRole { SYSTEM, USER, ASSISTANT, TOOL } +/** + * Budget class of an entry. REQUIRED entries (task, constraints, loaded inputs, failures) are + * never lossily compressed or dropped — if they alone exceed the budget the build fails fast + * instead of silently trimming. OPTIONAL entries (repo map, retrieval, journal, profiles) share + * whatever budget remains, each source type under its own ceiling. + */ +@Serializable +enum class ContextBucket { REQUIRED, OPTIONAL } + @Serializable data class ContextEntry( val id: ContextEntryId, @@ -20,4 +29,7 @@ data class ContextEntry( // Entries built outside that builder (e.g. router chat) leave this at 0 and fall // back to layer-priority ordering in PromptRenderer. val ordinal: Int = 0, + // Default OPTIONAL keeps prior behavior for callers that don't stamp buckets (router chat); + // the orchestrator stamps REQUIRED on task/constraint/input/failure entries. + val bucket: ContextBucket = ContextBucket.OPTIONAL, ) diff --git a/core/context/src/main/kotlin/com/correx/core/context/model/OptionalCeilings.kt b/core/context/src/main/kotlin/com/correx/core/context/model/OptionalCeilings.kt new file mode 100644 index 00000000..876a9afe --- /dev/null +++ b/core/context/src/main/kotlin/com/correx/core/context/model/OptionalCeilings.kt @@ -0,0 +1,30 @@ +package com.correx.core.context.model + +import kotlinx.serialization.Serializable + +/** + * Per-sourceType token ceilings for OPTIONAL bucket entries. Defaults are tuned for the + * 9–26B local-model tier running a 16k prompt budget: the optional block as a whole stays + * under ~5k tokens so the required block (task, constraints, loaded inputs, failures) and + * the running transcript always dominate the window. + * + * A source type absent from [caps] has no per-type ceiling — it is still bounded by the + * remaining post-required budget like everything else. + */ +@Serializable +data class OptionalCeilings( + val caps: Map = DEFAULT_CAPS, +) { + fun capFor(sourceType: String): Int? = caps[sourceType] + + companion object { + val DEFAULT_CAPS: Map = mapOf( + "repoMap" to 1200, + "relevantFiles" to 1600, + "decisionJournal" to 800, + "operatorProfile" to 400, + "projectProfile" to 600, + "artifactKindVocabulary" to 400, + ) + } +} diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/DecisionJournalRenderer.kt b/core/journal/src/main/kotlin/com/correx/core/journal/DecisionJournalRenderer.kt index 3d100186..6c6189f7 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/DecisionJournalRenderer.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/DecisionJournalRenderer.kt @@ -1,10 +1,13 @@ package com.correx.core.journal import com.correx.core.journal.model.DecisionJournalState +import com.correx.core.journal.model.DecisionKind +import com.correx.core.journal.model.DecisionRecord class DecisionJournalRenderer(private val keepLast: Int = 40) { fun render(state: DecisionJournalState, summaryText: String? = null): String { if (state.records.isEmpty() && summaryText == null) return "" + val visible = foldResolvedRetries(state.records) return buildString { appendLine("## Decision history (chronological — ground truth)") if (summaryText != null) { @@ -13,7 +16,34 @@ class DecisionJournalRenderer(private val keepLast: Int = 40) { appendLine("+${state.lowSalienceOmittedCount} routine transitions/retries omitted") } } - state.records.takeLast(keepLast).forEach { appendLine("- [${it.kind}] ${it.summary}") } + visible.takeLast(keepLast).forEach { appendLine("- [${it.kind}] ${it.summary}") } }.trimEnd() } + + /** + * A RETRY record is resolved once a later TRANSITION leaves its stage — the failure was + * overcome and its details are noise for every subsequent stage. Fold each resolved stage's + * retries into one summary line at the position of the first retry; unresolved (still-live) + * retries render verbatim. Render-time only: the journal state and event log are untouched. + */ + private fun foldResolvedRetries(records: List): List { + val resolvedAt = records + .filter { it.kind == DecisionKind.TRANSITION && it.fromStageId != null } + .groupBy { it.fromStageId!! } + .mapValues { (_, transitions) -> transitions.maxOf { it.sequence } } + val foldedStages = mutableSetOf() + return records.mapNotNull { record -> + val stage = record.stageId + val resolution = stage?.let { resolvedAt[it] } + if (record.kind != DecisionKind.RETRY || resolution == null || record.sequence > resolution) { + return@mapNotNull record + } + if (stage in foldedStages) return@mapNotNull null + foldedStages += stage + val count = records.count { + it.kind == DecisionKind.RETRY && it.stageId == stage && it.sequence <= resolution + } + record.copy(summary = "$stage: $count retr${if (count == 1) "y" else "ies"}, resolved — stage completed") + } + } } diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt b/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt index 9aa6b08b..e12a990c 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt @@ -53,6 +53,7 @@ class DefaultDecisionJournalReducer : DecisionJournalReducer { kind = DecisionKind.TRANSITION, summary = "Advanced ${p.from.value} → ${p.to.value} (${p.transitionId.value})", stageId = p.to.value, + fromStageId = p.from.value, ) is RetryAttemptedEvent -> DecisionRecord( sequence = event.sessionSequence, diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt index 231d5332..67f08901 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt @@ -25,4 +25,8 @@ data class DecisionRecord( val kind: DecisionKind, val summary: String, val stageId: String? = null, + // TRANSITION only: the stage the workflow advanced FROM. Lets the renderer tell that a + // stage's earlier RETRY records are resolved (the stage was eventually left successfully) + // and fold them to a single summary line. + val fromStageId: String? = null, ) diff --git a/core/journal/src/test/kotlin/com/correx/core/journal/DecisionJournalRendererTest.kt b/core/journal/src/test/kotlin/com/correx/core/journal/DecisionJournalRendererTest.kt index 41a9c51d..3a5c1261 100644 --- a/core/journal/src/test/kotlin/com/correx/core/journal/DecisionJournalRendererTest.kt +++ b/core/journal/src/test/kotlin/com/correx/core/journal/DecisionJournalRendererTest.kt @@ -77,4 +77,44 @@ class DecisionJournalRendererTest { assertTrue(result.contains("+2 routine transitions/retries omitted")) assertFalse(result.contains("- [")) } + + @Test + fun `resolved retries fold to a single summary line`() { + val state = DecisionJournalState( + records = listOf( + record(1, DecisionKind.RETRY, "Retry 1/3 of scaffold: did not produce").copy(stageId = "scaffold"), + record(2, DecisionKind.RETRY, "Retry 2/3 of scaffold: did not produce").copy(stageId = "scaffold"), + record(3, DecisionKind.TRANSITION, "Advanced scaffold → hook (t1)") + .copy(stageId = "hook", fromStageId = "scaffold"), + ), + ) + val result = renderer.render(state) + assertFalse(result.contains("did not produce"), "resolved retry detail must not render") + assertTrue(result.contains("- [RETRY] scaffold: 2 retries, resolved — stage completed")) + assertTrue(result.contains("- [TRANSITION] Advanced scaffold → hook")) + } + + @Test + fun `unresolved retries render verbatim`() { + val state = DecisionJournalState( + records = listOf( + record(1, DecisionKind.RETRY, "Retry 1/3 of hook: did not produce").copy(stageId = "hook"), + ), + ) + val result = renderer.render(state) + assertTrue(result.contains("- [RETRY] Retry 1/3 of hook: did not produce")) + } + + @Test + fun `retries after the stage was last left are not folded`() { + val state = DecisionJournalState( + records = listOf( + record(1, DecisionKind.TRANSITION, "Advanced scaffold → hook (t1)") + .copy(stageId = "hook", fromStageId = "scaffold"), + record(2, DecisionKind.RETRY, "Retry 1/3 of scaffold: regression").copy(stageId = "scaffold"), + ), + ) + val result = renderer.render(state) + assertTrue(result.contains("- [RETRY] Retry 1/3 of scaffold: regression")) + } } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt index 622cebc7..20afac3b 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ContextFeedback.kt @@ -73,7 +73,7 @@ fun buildArtifactKindVocabularyEntry(kinds: Collection): ContextEn fun buildRelevantFilesEntry(hits: List): ContextEntry { val content = buildString { append("## Relevant files (retrieved by semantic similarity)\n") - hits.forEach { append("- ").append(it.text).append("\n") } + hits.map { it.text }.distinct().forEach { append("- ").append(it).append("\n") } }.trimEnd() return ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 09a96a0d..80782435 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -20,6 +20,8 @@ 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 @@ -29,7 +31,11 @@ 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.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 @@ -128,6 +134,8 @@ 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 @@ -165,6 +173,31 @@ private const val READ_BEFORE_WRITE_CODE = "READ_BEFORE_WRITE" private const val REFERENCE_EXISTS_CODE = "REFERENCE_EXISTS" private const val OUTPUT_SUMMARY_LIMIT = 500 private const val REPO_MAP_INJECT_TOP_K = 30 +private const val REPO_MAP_FILES_PER_DIR = 8 + +// A stage prompt must explicitly ask about documentation for .md/docs paths to enter the repo +// layout listing; otherwise docs only reach context through a real retrieval hit. +private val DOCS_REQUEST_REGEX = + Regex("""\b(docs?|readme|adr|changelog|documentation|markdown)\b|\.md\b""", RegexOption.IGNORE_CASE) + +private fun isDocPath(path: String): Boolean = + path.endsWith(".md", ignoreCase = true) || + path.split('/').any { it.equals("docs", ignoreCase = true) || it.equals("doc", ignoreCase = true) } + +// Source types whose entries land in the REQUIRED context bucket (see stampBuckets). +private val REQUIRED_SOURCE_TYPES = setOf( + "systemPrompt", + "agentPrompt", + "agentInstructions", + "claimedTask", + "schemaInstruction", + "steeringNote", + "clarificationAnswer", + "rejectionFeedback", + "retryFeedback", + "neededArtifact", + "criticFeedback", +) // HTTP statuses that are transient despite being 4xx (F-002 retry classification). private const val HTTP_TIMEOUT = 408 @@ -236,6 +269,10 @@ abstract class SessionOrchestrator( * Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */ protected val artifactContentCache: ConcurrentHashMap = ConcurrentHashMap() + /** Deterministic extraction/repair ladder for near-miss LLM artifact text (prose-wrapped JSON, + * code fences, trailing commas). Pure — recomputes on replay, records no events. */ + private val artifactExtractionPipeline = ArtifactExtractionPipeline() + abstract suspend fun run( sessionId: SessionId, graph: WorkflowGraph, @@ -452,18 +489,27 @@ abstract class SessionOrchestrator( } else { emptyList() } - var accumulatedEntries = + var accumulatedEntries = stampBuckets( systemPrompt + profileEntries + projectProfileEntries + agentInstructionsEntries + journalEntries + repoMapEntries + claimedTaskEntries + needsEntries + schemaEntries + vocabularyEntries + promptEntries + steeringEntries + - clarificationEntries + retryFeedbackEntries - val contextPack = contextPackBuilder.build( - id = ContextPackId(UUID.randomUUID().toString()), - sessionId = sessionId, - stageId = stageId, - entries = accumulatedEntries, - budget = TokenBudget(limit = stageConfig.tokenBudget), + clarificationEntries + retryFeedbackEntries, ) + 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 @@ -651,11 +697,36 @@ abstract class SessionOrchestrator( // (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 artifactText = llmArtifactOverride ?: inferenceResult.response.text - val artifactHash = llmArtifactOverride - ?.let { artifactStore.put(it.toByteArray()) } - ?: inferenceResult.responseArtifactId - llmEmittedSlots.firstOrNull()?.let { slot -> + 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( @@ -690,6 +761,122 @@ abstract class SessionOrchestrator( } } + private sealed interface ArtifactLadderOutcome { + data class Text(val text: String) : ArtifactLadderOutcome + data class Reject(val failure: StageExecutionResult.Failure) : ArtifactLadderOutcome + } + + /** + * 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. + */ + private suspend fun 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. */ + private suspend fun 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 + } + + private suspend fun emitArtifactRepairAttempted( + sessionId: SessionId, + stageId: StageId, + slot: TypedArtifactSlot, + classification: ArtifactFailure, + rung: String, + ) = emit( + sessionId, + ArtifactRepairAttemptedEvent(slot.name, classification.name, rung, sessionId, stageId), + ) + + private suspend fun emitArtifactRepairResolved( + sessionId: SessionId, + stageId: StageId, + slot: TypedArtifactSlot, + json: String, + ) { + val hash = artifactStore.put(json.toByteArray()) + emit(sessionId, ArtifactRepairResolvedEvent(slot.name, hash, sessionId, stageId)) + } + private fun error( stageId: StageId, path: String, @@ -1347,22 +1534,32 @@ abstract class SessionOrchestrator( } /** - * Renders a top-K slice of the recorded repo map as a droppable L3 context entry. Reads - * the latest [RepoMapComputedEvent] from the log (replay-safe — never re-scans the FS). - * Not pinned: under budget pressure this yields before the working set (paths + symbols - * only, so the model can `file_read` for detail). + * 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). */ - private suspend fun buildRepoMapEntries(sessionId: SessionId): List { + private suspend fun buildRepoMapEntries(sessionId: SessionId, stagePrompt: String = ""): List { val map = eventStore.read(sessionId) .mapNotNull { it.payload as? RepoMapComputedEvent } .lastOrNull() ?: return emptyList() - val top = map.entries.sortedByDescending { it.score }.take(REPO_MAP_INJECT_TOP_K) - if (top.isEmpty()) return emptyList() + val 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 map (top files by recency — read files for detail)") - top.forEach { entry -> - val symbols = if (entry.symbols.isEmpty()) "" else ": ${entry.symbols.joinToString(", ")}" - appendLine("- ${entry.path}$symbols") + 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(REPO_MAP_FILES_PER_DIR) + val more = names.size - shown.size + val suffix = if (more > 0) ", …(+$more more)" else "" + appendLine("- $dir/ (${names.size}): ${shown.joinToString(", ")}$suffix") } }.trimEnd() return listOf( @@ -1396,15 +1593,15 @@ abstract class SessionOrchestrator( stageId: StageId, stageConfig: StageConfig, ): List { + val stagePrompt = (stageConfig.metadata["promptInline"] ?: stageConfig.metadata["prompt"] ?: stageId.value) + .trim().ifBlank { stageId.value } val recorded = eventStore.read(sessionId) .mapNotNull { it.payload as? RepoKnowledgeRetrievedEvent } .lastOrNull { it.stageId == stageId } - if (recorded != null) return repoEntriesOrMapFloor(sessionId, recorded.hits) + if (recorded != null) return repoEntriesOrMapFloor(sessionId, recorded.hits, stagePrompt) - val retriever = repoKnowledgeRetriever ?: return buildRepoMapEntries(sessionId) - val query = (stageConfig.metadata["promptInline"] ?: stageConfig.metadata["prompt"] ?: stageId.value) - .trim().ifBlank { stageId.value } - val hits = runCatching { retriever.retrieve(sessionId, query, REPO_MAP_INJECT_TOP_K) } + val retriever = repoKnowledgeRetriever ?: return buildRepoMapEntries(sessionId, stagePrompt) + val hits = runCatching { retriever.retrieve(sessionId, stagePrompt, REPO_MAP_INJECT_TOP_K) } .getOrElse { e -> if (e is CancellationException) throw e log.warn("repo-knowledge retrieval failed for stage {}: {}", stageId.value, e.message) @@ -1412,8 +1609,8 @@ abstract class SessionOrchestrator( } // 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, query, hits)) - return repoEntriesOrMapFloor(sessionId, hits) + if (hits.isNotEmpty()) emit(sessionId, RepoKnowledgeRetrievedEvent(sessionId, stageId, stagePrompt, hits)) + return repoEntriesOrMapFloor(sessionId, hits, stagePrompt) } /** @@ -1424,9 +1621,25 @@ abstract class SessionOrchestrator( private suspend fun repoEntriesOrMapFloor( sessionId: SessionId, hits: List, + stagePrompt: String, ): List { val useful = hits.filter { it.score > 0f } - return if (useful.isEmpty()) buildRepoMapEntries(sessionId) else listOf(buildRelevantFilesEntry(useful)) + 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. + */ + private fun stampBuckets(entries: List): List = entries.map { + if (it.sourceType in REQUIRED_SOURCE_TYPES) it.copy(bucket = ContextBucket.REQUIRED) else it } private suspend fun buildNeedsArtifactEntries( @@ -1443,7 +1656,13 @@ abstract class SessionOrchestrator( "This is reviewer feedback on your previous attempt. " + "Address every finding before completing the stage.\n$cached" } else { - "## Input artifact: ${needed.value}\n$cached" + // 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()), @@ -1457,6 +1676,38 @@ abstract class SessionOrchestrator( } } + private fun 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). + */ + private fun 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() + } + private suspend fun emitToolArtifacts( sessionId: SessionId, stageId: StageId, diff --git a/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt b/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt index 51f4006e..497cb24b 100644 --- a/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt +++ b/testing/deterministic/src/test/kotlin/DefaultContextPackBuilderTest.kt @@ -1,6 +1,8 @@ import com.correx.core.context.builder.DefaultContextPackBuilder import com.correx.core.context.builder.DefaultDecisionPointBuilder import com.correx.core.context.compression.DefaultContextCompressor +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.TokenBudget @@ -143,4 +145,50 @@ class DefaultContextPackBuilderTest { // ordinals are stamped from input order assertEquals(listOf(1, 2, 3, 4), pack.layers[ContextLayer.L2]!!.map { it.ordinal }) } + + @Test + fun `required entries survive intact under budget pressure`() = kotlinx.coroutines.runBlocking { + val entries = listOf( + entry("task", ContextLayer.L1, 400).copy(sourceType = "agentPrompt", bucket = ContextBucket.REQUIRED), + entry("input", ContextLayer.L1, 400).copy(sourceType = "neededArtifact", bucket = ContextBucket.REQUIRED), + entry("map", ContextLayer.L3, 5000).copy(sourceType = "repoMap"), + ) + val pack = builder.build(packId, sessionId, stageId, entries, TokenBudget(limit = 1000)) + val retained = pack.layers.values.flatten() + assertTrue(retained.any { it.id == ContextEntryId("task") }, "required task must be retained") + assertTrue(retained.any { it.id == ContextEntryId("input") }, "required input must be retained") + // and retained uncompressed — same token estimate as input + assertEquals(400, retained.first { it.id == ContextEntryId("task") }.tokenEstimate) + } + + @Test + fun `build fails fast when required bucket alone exceeds the budget`() { + val entries = listOf( + entry("task", ContextLayer.L1, 900).copy(sourceType = "agentPrompt", bucket = ContextBucket.REQUIRED), + entry("input", ContextLayer.L1, 900).copy(sourceType = "neededArtifact", bucket = ContextBucket.REQUIRED), + ) + val thrown = org.junit.jupiter.api.assertThrows { + kotlinx.coroutines.runBlocking { + builder.build(packId, sessionId, stageId, entries, TokenBudget(limit = 1000)) + } + } + assertEquals(1800, thrown.requiredTokens) + assertEquals(1000, thrown.budgetLimit) + assertEquals(900, thrown.breakdown["agentPrompt"]) + } + + @Test + fun `optional groups are capped by their per-type ceiling`() = kotlinx.coroutines.runBlocking { + // Journal ceiling is 800 by default: a 5000-token journal must not survive whole even + // though the overall budget would fit it. + val entries = listOf( + entry("task", ContextLayer.L1, 100).copy(sourceType = "agentPrompt", bucket = ContextBucket.REQUIRED), + entry("journal", ContextLayer.L0, 5000).copy(sourceType = "decisionJournal"), + ) + val pack = builder.build(packId, sessionId, stageId, entries, TokenBudget(limit = 16384)) + val journalTokens = pack.layers.values.flatten() + .filter { it.sourceType == "decisionJournal" } + .sumOf { it.tokenEstimate } + assertTrue(journalTokens <= 800, "journal must be capped to 800 tokens, was $journalTokens") + } } diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index c044c389..39f2454d 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -423,6 +423,28 @@ class SessionOrchestratorIntegrationTest { val sessionId = SessionId("s-trunc") val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + // Seed a recorded repo map so the stage context carries an OPTIONAL entry the dropping + // compressor can discard: required entries (like the system prompt) are pinned by the + // bucket split and never reach the compressor. + eventStore.append( + com.correx.core.events.events.NewEvent( + com.correx.core.events.events.EventMetadata( + com.correx.core.events.types.EventId("rm-trunc"), + sessionId, + kotlinx.datetime.Instant.parse("2026-06-04T00:00:00Z"), + 1, + null, + null, + ), + com.correx.core.events.events.RepoMapComputedEvent( + sessionId, + "/repo", + listOf(com.correx.core.events.events.RepoMapEntry("src/Main.kt", 0.9)), + kotlinx.datetime.Instant.parse("2026-06-04T00:00:00Z"), + ), + ), + ) + truncOrchestrator.run(sessionId, truncGraph, config) val truncated = eventStore.read(sessionId).mapNotNull { it.payload as? ContextTruncatedEvent }