From b976a5c92ac2989d32995d6ccdc2c15ae75240ec Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 2 Jun 2026 13:58:06 +0400 Subject: [PATCH] feat(kernel): emit ContextTruncatedEvent on workflow context overflow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit DefaultContextPackBuilder already enforced the token budget, but SessionOrchestrator discarded the entriesDropped signal at both build sites — overflow was enforced yet unobserved. Emit ContextTruncatedEvent when entriesDropped>0, matching the router path (open-threads 6.2 contract). --- .../orchestration/SessionOrchestrator.kt | 25 ++++++++++ .../SessionOrchestratorIntegrationTest.kt | 50 +++++++++++++++++++ 2 files changed, 75 insertions(+) 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 c15a7a04..cdd294c6 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 @@ -23,6 +23,7 @@ import com.correx.core.context.model.EntryRole import com.correx.core.events.events.ApprovalDecisionResolvedEvent import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.ContextTruncatedEvent import com.correx.core.events.events.ToolCallAssessedEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.ArtifactValidatingEvent @@ -267,6 +268,7 @@ abstract class SessionOrchestrator( entries = systemPrompt + schemaEntries + promptEntries + steeringEntries, budget = TokenBudget(limit = stageConfig.tokenBudget), ) + emitContextTruncationIfNeeded(sessionId, stageId, contextPack) var currentContext = contextPack var inferenceResult = runInference( @@ -310,6 +312,7 @@ abstract class SessionOrchestrator( entries = allEntries, budget = TokenBudget(limit = stageConfig.tokenBudget), ) + emitContextTruncationIfNeeded(sessionId, stageId, currentContext) inferenceResult = runInference( sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat, ) @@ -937,6 +940,28 @@ abstract class SessionOrchestrator( // --- event emission --- + // Surface context-budget overflow as an event on the workflow path. The builder already + // enforces the budget (drops oldest/compressible entries, pinning steering + event history); + // this records that it happened so replay and operators can see it, mirroring the router path. + private suspend fun emitContextTruncationIfNeeded( + sessionId: SessionId, + stageId: StageId, + contextPack: ContextPack, + ) { + val meta = contextPack.compressionMetadata + if (meta.entriesDropped <= 0) return + emit( + sessionId, + ContextTruncatedEvent( + sessionId = sessionId, + turnId = stageId.value, + entriesDropped = meta.entriesDropped, + truncatedLayers = meta.truncatedLayers.map { it.name }, + timestampMs = Clock.System.now().toEpochMilliseconds(), + ), + ) + } + internal suspend fun emit(sessionId: SessionId, payload: EventPayload) { log.debug("[session {}] emitting event, payload: {}", sessionId.value.take(7), payload) eventStore.append( diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index d25fb7c0..014d7048 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -10,7 +10,13 @@ import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.approvals.model.ApprovalScopeIdentity import com.correx.core.artifacts.DefaultArtifactReducer import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.context.builder.DefaultContextPackBuilder +import com.correx.core.context.compression.CompressionStrategy +import com.correx.core.context.compression.ContextCompressor +import com.correx.core.context.model.ContextEntry +import com.correx.core.context.model.TokenBudget import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.ContextTruncatedEvent import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.OrchestrationPausedEvent @@ -22,6 +28,7 @@ import com.correx.core.events.execution.RetryPolicy import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId +import com.correx.core.events.types.TransitionId import com.correx.core.inference.InferenceRepository import com.correx.core.inference.InferenceRouter import com.correx.core.inference.InferenceState @@ -39,6 +46,7 @@ import com.correx.core.sessions.ApprovalMode import com.correx.core.sessions.DefaultSessionRepository import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.core.sessions.projections.replay.EventReplayer +import com.correx.core.transitions.evaluation.PromptResolver import com.correx.core.transitions.graph.StageConfig import com.correx.core.transitions.graph.TransitionEdge import com.correx.core.transitions.graph.WorkflowGraph @@ -304,6 +312,48 @@ class SessionOrchestratorIntegrationTest { assertNotNull(failed.reason) } + @Test + fun `context budget overflow on the workflow path emits ContextTruncatedEvent`(): Unit = runBlocking { + // A compressor that drops everything forces the builder to report dropped entries; the + // orchestrator must surface that as a ContextTruncatedEvent (mirroring the router path). + val droppingBuilder = DefaultContextPackBuilder( + object : ContextCompressor { + override fun compress( + entries: List, + budget: TokenBudget, + strategy: CompressionStrategy, + ): List = emptyList() + }, + ) + val truncGraph = WorkflowGraph( + id = "trunc-test", + stages = mapOf( + StageId("A") to StageConfig(metadata = mapOf("systemPrompt" to "you are a helpful assistant")), + ), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + val truncOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + // PromptResolver { it } makes the systemPrompt metadata resolve to non-blank text, + // so there is a compressible entry for the dropping compressor to discard. + engines = engines.copy(contextPackBuilder = droppingBuilder, promptResolver = PromptResolver { it }), + retryCoordinator = retryCoordinator, + artifactStore = artifactStore, + ) + val sessionId = SessionId("s-trunc") + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + + truncOrchestrator.run(sessionId, truncGraph, config) + + val truncated = eventStore.read(sessionId).mapNotNull { it.payload as? ContextTruncatedEvent } + assertTrue(truncated.isNotEmpty(), "expected a ContextTruncatedEvent when the budget drops entries") + assertEquals("A", truncated.first().turnId) + assertTrue(truncated.first().entriesDropped >= 1) + } + @Test fun `artifactStore put is called for prompt and response and ids appear on inference events`(): Unit = runBlocking { val recordingStore = RecordingArtifactStore()