feat(context): required/optional bucket split, budget ceilings, grounded handoffs

Context rework from the 2026-07-05 context-poisoning diagnosis (session
6ca236c4 read-loop):

- ContextBucket on ContextEntry: REQUIRED (task, constraints, needed
  artifacts, failure feedback) is never pruned/compressed/dropped; the
  build fails fast (RequiredContextOverflowException -> non-retryable
  stage failure) if required alone exceeds the stage budget.
- OptionalCeilings: per-sourceType token caps for the optional block
  (repoMap 1200, relevantFiles 1600, journal 800, profiles 400/600,
  vocabulary 400), tuned for the 9-26B tier at 16k; enforced by
  truncation (journal keeps its tail, others their head).
- Repo map is now a structural 'what exists' listing (alphabetical,
  per-directory, no recency ranking, no symbols); .md/docs paths are
  gated behind an explicit docs mention in the stage prompt — retrieval
  hits remain the only other path for docs to enter context.
- file_written needs render as a manifest of ALL files the producing
  stage wrote (projected from ArtifactContentStored -> ToolInvocation
  -> FileWritten events), fixing the last-write-wins collapse; content
  stays lazy behind file_read.
- Decision journal folds a stage's RETRY records into one summary line
  once a later transition leaves that stage (render-time only).
- Dedupe retrieved-file lines in buildRelevantFilesEntry.
This commit is contained in:
2026-07-05 18:04:49 +04:00
parent f90f2ab39d
commit 33ea44d8cc
12 changed files with 545 additions and 40 deletions
@@ -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<ContextEntry>): List<ContextEntry> {
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 }
}
@@ -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<String, Int>,
) : IllegalStateException(
"Required context ($requiredTokens tokens) exceeds the stage budget ($budgetLimit); " +
"by sourceType: " + breakdown.entries.joinToString(", ") { "${it.key}=${it.value}" },
)
@@ -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,
)
@@ -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
* 926B 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<String, Int> = DEFAULT_CAPS,
) {
fun capFor(sourceType: String): Int? = caps[sourceType]
companion object {
val DEFAULT_CAPS: Map<String, Int> = mapOf(
"repoMap" to 1200,
"relevantFiles" to 1600,
"decisionJournal" to 800,
"operatorProfile" to 400,
"projectProfile" to 600,
"artifactKindVocabulary" to 400,
)
}
}
@@ -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<DecisionRecord>): List<DecisionRecord> {
val resolvedAt = records
.filter { it.kind == DecisionKind.TRANSITION && it.fromStageId != null }
.groupBy { it.fromStageId!! }
.mapValues { (_, transitions) -> transitions.maxOf { it.sequence } }
val foldedStages = mutableSetOf<String>()
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")
}
}
}
@@ -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,
@@ -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,
)
@@ -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"))
}
}
@@ -73,7 +73,7 @@ fun buildArtifactKindVocabularyEntry(kinds: Collection<ArtifactKind>): ContextEn
fun buildRelevantFilesEntry(hits: List<RepoKnowledgeHit>): 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()),
@@ -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<String, String> = 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<ContextEntry> {
private suspend fun buildRepoMapEntries(sessionId: SessionId, stagePrompt: String = ""): List<ContextEntry> {
val map = eventStore.read(sessionId)
.mapNotNull { it.payload as? RepoMapComputedEvent }
.lastOrNull() ?: return emptyList()
val top = map.entries.sortedByDescending { it.score }.take(REPO_MAP_INJECT_TOP_K)
if (top.isEmpty()) return emptyList()
val 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<ContextEntry> {
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<RepoKnowledgeHit>,
stagePrompt: String,
): List<ContextEntry> {
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<ContextEntry>): List<ContextEntry> = 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,