From 1a1b5cc5ed99ad43803433d7b3933ff19fdd8d70 Mon Sep 17 00:00:00 2001 From: kami Date: Sat, 20 Jun 2026 10:53:41 +0000 Subject: [PATCH] feat(kernel): stage-level plan checkpointing (C-A2) - StageCheckpointPassed/FailedEvent (registered + serialization test): per-stage reconciliation of produced artifacts vs the locked plan's produces-slots - StageCheckpointReconciler (pure) + tests - emitStageCheckpoint wired in after verifyProduces, runs regardless of its pass/fail (verifyProduces still owns the halt); gated on an ExecutionPlanLocked in the session log so non-plan workflows are unaffected Co-Authored-By: Claude Opus 4.8 --- .../events/events/StageCheckpointEvents.kt | 42 ++++++++++++++++ .../events/serialization/Serialization.kt | 4 ++ .../StageCheckpointEventSerializationTest.kt | 42 ++++++++++++++++ .../orchestration/SessionOrchestrator.kt | 33 ++++++++++++- .../StageCheckpointReconciler.kt | 23 +++++++++ .../StageCheckpointReconcilerTest.kt | 49 +++++++++++++++++++ 6 files changed, 192 insertions(+), 1 deletion(-) create mode 100644 core/events/src/main/kotlin/com/correx/core/events/events/StageCheckpointEvents.kt create mode 100644 core/events/src/test/kotlin/com/correx/core/events/serialization/StageCheckpointEventSerializationTest.kt create mode 100644 core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconciler.kt create mode 100644 core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconcilerTest.kt diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/StageCheckpointEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/StageCheckpointEvents.kt new file mode 100644 index 00000000..e0e8cb6a --- /dev/null +++ b/core/events/src/main/kotlin/com/correx/core/events/events/StageCheckpointEvents.kt @@ -0,0 +1,42 @@ +package com.correx.core.events.events + +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +/** + * Per-stage reconciliation of a stage's produced artifacts against the locked plan's produces-slots, + * recorded so plan adherence is observable and replayable. Emitted only on plan-driven (phase-2) + * runs — a session whose log contains an [ExecutionPlanLockedEvent]; non-plan workflows emit nothing. + * + * The *passed* variant means every expected slot was produced: [producedArtifacts] is a superset of + * [expectedArtifacts] (extra artifacts beyond the expected set do not fail the checkpoint). + */ +@Serializable +@SerialName("StageCheckpointPassed") +data class StageCheckpointPassedEvent( + val sessionId: SessionId, + val stageId: StageId, + val expectedArtifacts: Set, + val producedArtifacts: Set, +) : EventPayload + +/** + * Per-stage reconciliation of a stage's produced artifacts against the locked plan's produces-slots, + * recorded so plan adherence is observable and replayable. Emitted only on plan-driven (phase-2) + * runs — a session whose log contains an [ExecutionPlanLockedEvent]; non-plan workflows emit nothing. + * + * The *failed* variant accompanies the stage halt owned by `verifyProduces` and surfaces *why* the + * plan diverged: [missingArtifacts] are the expected slots ([expectedArtifacts] minus + * [producedArtifacts]) the stage never produced. + */ +@Serializable +@SerialName("StageCheckpointFailed") +data class StageCheckpointFailedEvent( + val sessionId: SessionId, + val stageId: StageId, + val expectedArtifacts: Set, + val producedArtifacts: Set, + val missingArtifacts: Set, +) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt index 08076475..32fb6e6a 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt @@ -45,6 +45,8 @@ import com.correx.core.events.events.RepoMapComputedEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.WorkspaceStateObservedEvent import com.correx.core.events.events.RiskAssessedEvent +import com.correx.core.events.events.StageCheckpointFailedEvent +import com.correx.core.events.events.StageCheckpointPassedEvent import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.PreemptRedirectBlockedEvent import com.correx.core.events.events.PreemptRedirectEvent @@ -131,6 +133,8 @@ val eventModule = SerializersModule { subclass(IdeaCapturedEvent::class) subclass(IdeaDiscardedEvent::class) subclass(CritiqueOutcomeCorrelatedEvent::class) + subclass(StageCheckpointPassedEvent::class) + subclass(StageCheckpointFailedEvent::class) } } diff --git a/core/events/src/test/kotlin/com/correx/core/events/serialization/StageCheckpointEventSerializationTest.kt b/core/events/src/test/kotlin/com/correx/core/events/serialization/StageCheckpointEventSerializationTest.kt new file mode 100644 index 00000000..d84db0a5 --- /dev/null +++ b/core/events/src/test/kotlin/com/correx/core/events/serialization/StageCheckpointEventSerializationTest.kt @@ -0,0 +1,42 @@ +package com.correx.core.events.serialization + +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.StageCheckpointFailedEvent +import com.correx.core.events.events.StageCheckpointPassedEvent +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.assertTrue + +class StageCheckpointEventSerializationTest { + + @Test + fun `StageCheckpointPassedEvent round-trips as polymorphic EventPayload`() { + val sample: EventPayload = StageCheckpointPassedEvent( + sessionId = SessionId("s"), + stageId = StageId("plan"), + expectedArtifacts = setOf("plan.json"), + producedArtifacts = setOf("plan.json", "notes.md"), + ) + val encoded = eventJson.encodeToString(EventPayload.serializer(), sample) + assertTrue(encoded.contains("\"type\":\"StageCheckpointPassed\""), "SerialName must be present: $encoded") + val decoded = eventJson.decodeFromString(EventPayload.serializer(), encoded) + assertEquals(sample, decoded) + } + + @Test + fun `StageCheckpointFailedEvent round-trips as polymorphic EventPayload`() { + val sample: EventPayload = StageCheckpointFailedEvent( + sessionId = SessionId("s"), + stageId = StageId("plan"), + expectedArtifacts = setOf("plan.json", "diff.patch"), + producedArtifacts = setOf("plan.json"), + missingArtifacts = setOf("diff.patch"), + ) + val encoded = eventJson.encodeToString(EventPayload.serializer(), sample) + assertTrue(encoded.contains("\"type\":\"StageCheckpointFailed\""), "SerialName must be present: $encoded") + val decoded = eventJson.decodeFromString(EventPayload.serializer(), encoded) + assertEquals(sample, decoded) + } +} 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 85bca6c1..3d3d946b 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 @@ -53,6 +53,8 @@ import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RiskAssessedEvent import com.correx.core.events.events.ExecutionPlanLockedEvent import com.correx.core.events.events.FileWrittenEvent +import com.correx.core.events.events.StageCheckpointFailedEvent +import com.correx.core.events.events.StageCheckpointPassedEvent import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.ToolExecutionCompletedEvent import com.correx.core.events.events.ToolExecutionFailedEvent @@ -549,7 +551,9 @@ abstract class SessionOrchestrator( is StageExecutionResult.Success -> { emitToolArtifacts(sessionId, stageId, stageConfig) emitLlmArtifacts(sessionId, stageId, stageConfig) - when (val produced = verifyProduces(sessionId, stageId, stageConfig)) { + val produced = verifyProduces(sessionId, stageId, stageConfig) + emitStageCheckpoint(sessionId, stageId, stageConfig) + when (produced) { is StageExecutionResult.Success -> when (val grounded = groundBriefReferences(sessionId, stageId, stageConfig, effectives)) { is StageExecutionResult.Success -> @@ -1360,6 +1364,33 @@ abstract class SessionOrchestrator( ) } + /** + * Stage-level plan checkpointing (BACKLOG §C-A2): make plan adherence observable and replayable + * by emitting a first-class checkpoint event per stage, reconciling produced-vs-expected against + * the confirmed (locked) plan. Only runs on plan-driven (phase-2) sessions — i.e. a log that + * holds an [ExecutionPlanLockedEvent]; non-plan workflows emit nothing. The halt itself is owned + * by [verifyProduces]; this method only EMITS — a [StageCheckpointFailedEvent] accompanies that + * halt and surfaces *why* the plan diverged. + */ + private suspend fun emitStageCheckpoint( + sessionId: SessionId, + stageId: StageId, + stageConfig: StageConfig, + ) { + if (eventStore.read(sessionId).none { it.payload is ExecutionPlanLockedEvent }) return + val expected = stageConfig.produces.map { it.name.value }.toSet() + if (expected.isEmpty()) return + val produced = eventStore.read(sessionId) + .mapNotNull { (it.payload as? ArtifactCreatedEvent)?.artifactId?.value } + .toSet() + val cp = StageCheckpointReconciler.reconcile(expected, produced) + if (cp.passed) { + emit(sessionId, StageCheckpointPassedEvent(sessionId, stageId, expected, produced)) + } else { + emit(sessionId, StageCheckpointFailedEvent(sessionId, stageId, expected, produced, cp.missing)) + } + } + /** * Entity grounding (role-reliability §3): for a stage that opted in (`ground_references`), * check every workspace-relative file path its brief named against the filesystem. Existence diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconciler.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconciler.kt new file mode 100644 index 00000000..de206782 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconciler.kt @@ -0,0 +1,23 @@ +package com.correx.core.kernel.orchestration + +/** + * The reconciliation of a stage's [produced] artifacts against the [expected] produces-slots of the + * locked plan. [missing] is the expected slots that were never produced; [passed] holds when none + * are missing. Extra produced artifacts beyond [expected] do not fail the checkpoint. + */ +internal data class StageCheckpoint( + val expected: Set, + val produced: Set, +) { + val missing: Set get() = expected - produced + val passed: Boolean get() = missing.isEmpty() +} + +/** + * Pure reconciler for stage-level plan checkpointing: compares the artifacts a stage produced + * against the artifacts the locked plan expected it to produce. Side-effect free and replay-safe. + */ +internal object StageCheckpointReconciler { + fun reconcile(expected: Set, produced: Set): StageCheckpoint = + StageCheckpoint(expected = expected, produced = produced) +} diff --git a/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconcilerTest.kt b/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconcilerTest.kt new file mode 100644 index 00000000..e325a166 --- /dev/null +++ b/core/kernel/src/test/kotlin/com/correx/core/kernel/orchestration/StageCheckpointReconcilerTest.kt @@ -0,0 +1,49 @@ +package com.correx.core.kernel.orchestration + +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class StageCheckpointReconcilerTest { + + @Test + fun `all expected produced - passed with no missing`() { + val cp = StageCheckpointReconciler.reconcile( + expected = setOf("plan.json", "diff.patch"), + produced = setOf("plan.json", "diff.patch"), + ) + assertTrue(cp.passed) + assertTrue(cp.missing.isEmpty()) + } + + @Test + fun `one missing - not passed with correct missing set`() { + val cp = StageCheckpointReconciler.reconcile( + expected = setOf("plan.json", "diff.patch"), + produced = setOf("plan.json"), + ) + assertFalse(cp.passed) + assertEquals(setOf("diff.patch"), cp.missing) + } + + @Test + fun `empty expected - passed`() { + val cp = StageCheckpointReconciler.reconcile( + expected = emptySet(), + produced = setOf("plan.json"), + ) + assertTrue(cp.passed) + assertTrue(cp.missing.isEmpty()) + } + + @Test + fun `extra produced beyond expected - still passed`() { + val cp = StageCheckpointReconciler.reconcile( + expected = setOf("plan.json"), + produced = setOf("plan.json", "notes.md", "scratch.txt"), + ) + assertTrue(cp.passed) + assertTrue(cp.missing.isEmpty()) + } +}