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 <noreply@anthropic.com>

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<String>,
+    val producedArtifacts: Set<String>,
+) : 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<String>,
+    val producedArtifacts: Set<String>,
+    val missingArtifacts: Set<String>,
+) : EventPayload

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)
     }
 }
 

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)
+    }
+}

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

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<String>,
+    val produced: Set<String>,
+) {
+    val missing: Set<String> 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<String>, produced: Set<String>): StageCheckpoint =
+        StageCheckpoint(expected = expected, produced = produced)
+}

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())
+    }
+}