feat(context): emit ContextAssembled manifest event per stage build (#307)
Root cause: context-assembly (unconfirmedFixEntries, promotedConceptEntries,
buildRetryFeedbackEntry, etc.) computes ContextEntry lists at stage-build time
but never records what was actually injected into a turn. The only way to
answer "did hint X fire in session Y" was decoding the CAS prompt artifact and
grepping raw JSON — the event log alone couldn't say.
Scope check: no existing event carries this (ContextTruncatedEvent only
reports drop counts, not the injected set) — added new ContextAssembledEvent
rather than extending an existing one.
Fix: ContextAssembledEvent (core/events/events/ContextEvents.kt) records ONE
manifest per stage's initial ContextPack build — sessionId, stageId,
contextPackId, and a List<ContextManifestEntry> of {sourceType, sourceId,
tokenEstimate, layer, role} pulled from the pack's actual (post-budget)
entries. No entry content — content stays a pure derived projection of the
event log (invariant #9/#6) and already lives in CAS. Registered in
eventModule (Serialization.kt). Emitted via
SessionOrchestrator.emitContextAssembled (SessionOrchestratorRepoContext.kt),
wired at the initial contextPackBuilder.build() call site in
SessionOrchestratorExecution.kt (mirrors emitContextTruncationIfNeeded).
Tests: ContextAssembledEventSerializationTest (round-trip + no-content-leak),
SessionOrchestratorIntegrationTest "stage context build emits
ContextAssembledEvent with a manifest (#307)".
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HgDL1v3GuQ9RZnYR6fDT95
This commit is contained in:
@@ -12,3 +12,37 @@ data class SteeringNoteAddedEvent(
|
|||||||
val content: String,
|
val content: String,
|
||||||
val stageId: StageId? = null,
|
val stageId: StageId? = null,
|
||||||
) : EventPayload
|
) : EventPayload
|
||||||
|
|
||||||
|
/**
|
||||||
|
* One entry in a [ContextAssembledEvent] manifest. Mirrors the identifying fields of
|
||||||
|
* core:context's ContextEntry (sourceType, sourceId, tokenEstimate, layer, role) but NEVER the
|
||||||
|
* content — content is a derived projection of the event log (reproducible on replay, see
|
||||||
|
* invariant #9 / #6) and already lives in CAS via the prompt artifact. This is manifest-only,
|
||||||
|
* for auditing what got injected without decoding CAS.
|
||||||
|
*/
|
||||||
|
@Serializable
|
||||||
|
data class ContextManifestEntry(
|
||||||
|
val sourceType: String,
|
||||||
|
val sourceId: String,
|
||||||
|
val tokenEstimate: Int,
|
||||||
|
val layer: String,
|
||||||
|
val role: String,
|
||||||
|
)
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Records the manifest of entries injected into a stage's initial context build (#307). Emitted
|
||||||
|
* once per [core.context.builder.ContextPackBuilder]-produced ContextPack, from the entries that
|
||||||
|
* actually made it into the pack (post budget/truncation) — the same set that
|
||||||
|
* [ContextTruncatedEvent] reports drops against. Purely observational: the hints themselves stay
|
||||||
|
* unevented derived projections; this only names what was fed to the model, so an operator can
|
||||||
|
* answer "did entry X fire in session Y" from the event log instead of CAS-spelunking.
|
||||||
|
*/
|
||||||
|
@Serializable
|
||||||
|
@SerialName("ContextAssembled")
|
||||||
|
data class ContextAssembledEvent(
|
||||||
|
val sessionId: SessionId,
|
||||||
|
val stageId: StageId,
|
||||||
|
val contextPackId: String,
|
||||||
|
val entries: List<ContextManifestEntry>,
|
||||||
|
val timestampMs: Long,
|
||||||
|
) : EventPayload
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ import com.correx.core.events.events.TalkieNarrationEvent
|
|||||||
import com.correx.core.events.events.OperatorProfileBoundEvent
|
import com.correx.core.events.events.OperatorProfileBoundEvent
|
||||||
import com.correx.core.events.events.ProjectProfileBoundEvent
|
import com.correx.core.events.events.ProjectProfileBoundEvent
|
||||||
import com.correx.core.events.events.SessionWorkspaceBoundEvent
|
import com.correx.core.events.events.SessionWorkspaceBoundEvent
|
||||||
|
import com.correx.core.events.events.ContextAssembledEvent
|
||||||
import com.correx.core.events.events.ContextTruncatedEvent
|
import com.correx.core.events.events.ContextTruncatedEvent
|
||||||
import com.correx.core.events.events.PossibleContradictionFlaggedEvent
|
import com.correx.core.events.events.PossibleContradictionFlaggedEvent
|
||||||
import com.correx.core.events.events.EgressHostsGrantedEvent
|
import com.correx.core.events.events.EgressHostsGrantedEvent
|
||||||
@@ -192,6 +193,7 @@ val eventModule = SerializersModule {
|
|||||||
subclass(AgentInstructionsBoundEvent::class)
|
subclass(AgentInstructionsBoundEvent::class)
|
||||||
subclass(L3MemoryRetrievedEvent::class)
|
subclass(L3MemoryRetrievedEvent::class)
|
||||||
subclass(ContextTruncatedEvent::class)
|
subclass(ContextTruncatedEvent::class)
|
||||||
|
subclass(ContextAssembledEvent::class)
|
||||||
subclass(ExecutionPlanLockedEvent::class)
|
subclass(ExecutionPlanLockedEvent::class)
|
||||||
subclass(ExecutionPlanRejectedEvent::class)
|
subclass(ExecutionPlanRejectedEvent::class)
|
||||||
subclass(PlanCompileCheckedEvent::class)
|
subclass(PlanCompileCheckedEvent::class)
|
||||||
|
|||||||
+55
@@ -0,0 +1,55 @@
|
|||||||
|
package com.correx.core.events.serialization
|
||||||
|
|
||||||
|
import com.correx.core.events.events.ContextAssembledEvent
|
||||||
|
import com.correx.core.events.events.ContextManifestEntry
|
||||||
|
import com.correx.core.events.events.EventPayload
|
||||||
|
import com.correx.core.events.types.SessionId
|
||||||
|
import com.correx.core.events.types.StageId
|
||||||
|
import kotlin.test.Test
|
||||||
|
import kotlin.test.assertEquals
|
||||||
|
import kotlin.test.assertFalse
|
||||||
|
|
||||||
|
class ContextAssembledEventSerializationTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `ContextAssembledEvent round-trips through eventModule`() {
|
||||||
|
val sample: EventPayload = ContextAssembledEvent(
|
||||||
|
sessionId = SessionId("s"),
|
||||||
|
stageId = StageId("implement"),
|
||||||
|
contextPackId = "pack-1",
|
||||||
|
entries = listOf(
|
||||||
|
ContextManifestEntry(
|
||||||
|
sourceType = "conceptPromotion",
|
||||||
|
sourceId = "concept-42",
|
||||||
|
tokenEstimate = 120,
|
||||||
|
layer = "L0",
|
||||||
|
role = "SYSTEM",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
timestampMs = 1_700_000_000_000L,
|
||||||
|
)
|
||||||
|
val encoded = eventJson.encodeToString(EventPayload.serializer(), sample)
|
||||||
|
assertEquals(sample, eventJson.decodeFromString(EventPayload.serializer(), encoded))
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `ContextAssembledEvent manifest never carries entry content`() {
|
||||||
|
val sample: EventPayload = ContextAssembledEvent(
|
||||||
|
sessionId = SessionId("s"),
|
||||||
|
stageId = StageId("implement"),
|
||||||
|
contextPackId = "pack-1",
|
||||||
|
entries = listOf(
|
||||||
|
ContextManifestEntry(
|
||||||
|
sourceType = "steering",
|
||||||
|
sourceId = "note-1",
|
||||||
|
tokenEstimate = 5,
|
||||||
|
layer = "L0",
|
||||||
|
role = "USER",
|
||||||
|
),
|
||||||
|
),
|
||||||
|
timestampMs = 0L,
|
||||||
|
)
|
||||||
|
val encoded = eventJson.encodeToString(EventPayload.serializer(), sample)
|
||||||
|
assertFalse(encoded.contains("\"content\""), "manifest must not carry entry content: $encoded")
|
||||||
|
}
|
||||||
|
}
|
||||||
+1
@@ -304,6 +304,7 @@ internal suspend fun SessionOrchestrator.executeStage(
|
|||||||
return StageExecutionResult.Failure(e.message ?: "required context overflow", retryable = false)
|
return StageExecutionResult.Failure(e.message ?: "required context overflow", retryable = false)
|
||||||
}
|
}
|
||||||
emitContextTruncationIfNeeded(sessionId, stageId, contextPack)
|
emitContextTruncationIfNeeded(sessionId, stageId, contextPack)
|
||||||
|
emitContextAssembled(sessionId, stageId, contextPack)
|
||||||
|
|
||||||
var currentContext = contextPack
|
var currentContext = contextPack
|
||||||
var inferenceResult = runInference(
|
var inferenceResult = runInference(
|
||||||
|
|||||||
+31
@@ -2,6 +2,8 @@ package com.correx.core.kernel.orchestration
|
|||||||
|
|
||||||
import com.correx.core.context.model.ContextPack
|
import com.correx.core.context.model.ContextPack
|
||||||
import com.correx.core.events.events.ApprovalRequestedEvent
|
import com.correx.core.events.events.ApprovalRequestedEvent
|
||||||
|
import com.correx.core.events.events.ContextAssembledEvent
|
||||||
|
import com.correx.core.events.events.ContextManifestEntry
|
||||||
import com.correx.core.events.events.ContextTruncatedEvent
|
import com.correx.core.events.events.ContextTruncatedEvent
|
||||||
import com.correx.core.events.events.InitialIntentEvent
|
import com.correx.core.events.events.InitialIntentEvent
|
||||||
import com.correx.core.events.events.OrchestrationPausedEvent
|
import com.correx.core.events.events.OrchestrationPausedEvent
|
||||||
@@ -78,6 +80,35 @@ internal suspend fun SessionOrchestrator.emitContextTruncationIfNeeded(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// #307: record the manifest of what got injected into the stage's initial context build — NOT
|
||||||
|
// the content (that's derivable/replayable, invariant #9, and already in CAS via the prompt
|
||||||
|
// artifact) — so the injected set is auditable from the event log alone.
|
||||||
|
internal suspend fun SessionOrchestrator.emitContextAssembled(
|
||||||
|
sessionId: SessionId,
|
||||||
|
stageId: StageId,
|
||||||
|
contextPack: ContextPack,
|
||||||
|
) {
|
||||||
|
val entries = contextPack.layers.values.flatten().map { entry ->
|
||||||
|
ContextManifestEntry(
|
||||||
|
sourceType = entry.sourceType,
|
||||||
|
sourceId = entry.sourceId,
|
||||||
|
tokenEstimate = entry.tokenEstimate,
|
||||||
|
layer = entry.layer.name,
|
||||||
|
role = entry.role.name,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
emit(
|
||||||
|
sessionId,
|
||||||
|
ContextAssembledEvent(
|
||||||
|
sessionId = sessionId,
|
||||||
|
stageId = stageId,
|
||||||
|
contextPackId = contextPack.id.value,
|
||||||
|
entries = entries,
|
||||||
|
timestampMs = Clock.System.now().toEpochMilliseconds(),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
internal fun SessionOrchestrator.fallbackTokenEstimate(content: String): Int {
|
internal fun SessionOrchestrator.fallbackTokenEstimate(content: String): Int {
|
||||||
return (content.length / 4).coerceAtLeast(1)
|
return (content.length / 4).coerceAtLeast(1)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ import com.correx.core.context.compression.ContextCompressor
|
|||||||
import com.correx.core.context.model.ContextEntry
|
import com.correx.core.context.model.ContextEntry
|
||||||
import com.correx.core.context.model.TokenBudget
|
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.ContextAssembledEvent
|
||||||
import com.correx.core.events.events.ContextTruncatedEvent
|
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
|
||||||
@@ -535,6 +536,22 @@ class SessionOrchestratorIntegrationTest {
|
|||||||
assertTrue(truncated.first().entriesDropped >= 1)
|
assertTrue(truncated.first().entriesDropped >= 1)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `stage context build emits ContextAssembledEvent with a manifest (#307)`(): Unit = runBlocking {
|
||||||
|
val sessionId = SessionId("s-manifest")
|
||||||
|
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0))
|
||||||
|
|
||||||
|
orchestrator.run(sessionId, graph, config)
|
||||||
|
|
||||||
|
val assembled = eventStore.read(sessionId).mapNotNull { it.payload as? ContextAssembledEvent }
|
||||||
|
assertTrue(assembled.isNotEmpty(), "expected a ContextAssembledEvent per stage context build")
|
||||||
|
val first = assembled.first()
|
||||||
|
assertEquals(StageId("A"), first.stageId)
|
||||||
|
assertTrue(first.entries.isNotEmpty(), "manifest should list the entries injected into the stage")
|
||||||
|
// Manifest-only: identifying fields present, never the content (content lives in CAS/replay).
|
||||||
|
assertTrue(first.entries.all { it.sourceType.isNotBlank() && it.layer.isNotBlank() && it.role.isNotBlank() })
|
||||||
|
}
|
||||||
|
|
||||||
@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