feat(kernel): emit ContextTruncatedEvent on workflow context overflow
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).
This commit is contained in:
+25
@@ -23,6 +23,7 @@ import com.correx.core.context.model.EntryRole
|
|||||||
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
||||||
import com.correx.core.events.events.ApprovalRequestedEvent
|
import com.correx.core.events.events.ApprovalRequestedEvent
|
||||||
import com.correx.core.events.events.ArtifactCreatedEvent
|
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.ToolCallAssessedEvent
|
||||||
import com.correx.core.events.events.ArtifactValidatedEvent
|
import com.correx.core.events.events.ArtifactValidatedEvent
|
||||||
import com.correx.core.events.events.ArtifactValidatingEvent
|
import com.correx.core.events.events.ArtifactValidatingEvent
|
||||||
@@ -267,6 +268,7 @@ abstract class SessionOrchestrator(
|
|||||||
entries = systemPrompt + schemaEntries + promptEntries + steeringEntries,
|
entries = systemPrompt + schemaEntries + promptEntries + steeringEntries,
|
||||||
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
||||||
)
|
)
|
||||||
|
emitContextTruncationIfNeeded(sessionId, stageId, contextPack)
|
||||||
|
|
||||||
var currentContext = contextPack
|
var currentContext = contextPack
|
||||||
var inferenceResult = runInference(
|
var inferenceResult = runInference(
|
||||||
@@ -310,6 +312,7 @@ abstract class SessionOrchestrator(
|
|||||||
entries = allEntries,
|
entries = allEntries,
|
||||||
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
budget = TokenBudget(limit = stageConfig.tokenBudget),
|
||||||
)
|
)
|
||||||
|
emitContextTruncationIfNeeded(sessionId, stageId, currentContext)
|
||||||
inferenceResult = runInference(
|
inferenceResult = runInference(
|
||||||
sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat,
|
sessionId, stageId, currentContext, stageConfig, config.stageTimeoutMs, responseFormat,
|
||||||
)
|
)
|
||||||
@@ -937,6 +940,28 @@ abstract class SessionOrchestrator(
|
|||||||
|
|
||||||
// --- event emission ---
|
// --- 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) {
|
internal suspend fun emit(sessionId: SessionId, payload: EventPayload) {
|
||||||
log.debug("[session {}] emitting event, payload: {}", sessionId.value.take(7), payload)
|
log.debug("[session {}] emitting event, payload: {}", sessionId.value.take(7), payload)
|
||||||
eventStore.append(
|
eventStore.append(
|
||||||
|
|||||||
@@ -10,7 +10,13 @@ import com.correx.core.approvals.model.ApprovalDecision
|
|||||||
import com.correx.core.approvals.model.ApprovalScopeIdentity
|
import com.correx.core.approvals.model.ApprovalScopeIdentity
|
||||||
import com.correx.core.artifacts.DefaultArtifactReducer
|
import com.correx.core.artifacts.DefaultArtifactReducer
|
||||||
import com.correx.core.artifactstore.ArtifactStore
|
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.ApprovalRequestedEvent
|
||||||
|
import com.correx.core.events.events.ContextTruncatedEvent
|
||||||
import com.correx.core.events.events.InferenceCompletedEvent
|
import com.correx.core.events.events.InferenceCompletedEvent
|
||||||
import com.correx.core.events.events.InferenceStartedEvent
|
import com.correx.core.events.events.InferenceStartedEvent
|
||||||
import com.correx.core.events.events.OrchestrationPausedEvent
|
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.ArtifactId
|
||||||
import com.correx.core.events.types.SessionId
|
import com.correx.core.events.types.SessionId
|
||||||
import com.correx.core.events.types.StageId
|
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.InferenceRepository
|
||||||
import com.correx.core.inference.InferenceRouter
|
import com.correx.core.inference.InferenceRouter
|
||||||
import com.correx.core.inference.InferenceState
|
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.DefaultSessionRepository
|
||||||
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
|
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
|
||||||
import com.correx.core.sessions.projections.replay.EventReplayer
|
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.StageConfig
|
||||||
import com.correx.core.transitions.graph.TransitionEdge
|
import com.correx.core.transitions.graph.TransitionEdge
|
||||||
import com.correx.core.transitions.graph.WorkflowGraph
|
import com.correx.core.transitions.graph.WorkflowGraph
|
||||||
@@ -304,6 +312,48 @@ class SessionOrchestratorIntegrationTest {
|
|||||||
assertNotNull(failed.reason)
|
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<ContextEntry>,
|
||||||
|
budget: TokenBudget,
|
||||||
|
strategy: CompressionStrategy,
|
||||||
|
): List<ContextEntry> = 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
|
@Test
|
||||||
fun `artifactStore put is called for prompt and response and ids appear on inference events`(): Unit = runBlocking {
|
fun `artifactStore put is called for prompt and response and ids appear on inference events`(): Unit = runBlocking {
|
||||||
val recordingStore = RecordingArtifactStore()
|
val recordingStore = RecordingArtifactStore()
|
||||||
|
|||||||
Reference in New Issue
Block a user