feat(workflow): ExecutionPlanCompiler + ExecutionPlanLockedEvent (Slice 3)

- ExecutionPlanLockedEvent: registered, replay-deterministic record of the locked plan (CAS ref + compiled stage/start ids). - ExecutionPlanCompiler: validated execution_plan JSON -> WorkflowGraph, reusing ConditionSpec.toCondition(); inline prompts via metadata[promptInline]; malformed plans throw WorkflowValidationException. - Replay test: locked event + plan JSON recompiles to the identical graph, no architect re-run, no FS reads.
2026-06-08 02:33:57 +04:00
parent be78eaa80f
commit 6c8c5e2ad9
8 changed files with 356 additions and 0 deletions
@@ -0,0 +1,22 @@
+package com.correx.core.events.events
+
+import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.types.SessionId
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+
+/**
+ * The architect's validated execution_plan was compiled into a runnable WorkflowGraph and
+ * locked as the baseline for phase 2. [planArtifactId] points at the validated plan JSON in
+ * CAS (referenced, not inlined); [stageIds]/[startStageId] pin the compiled shape so replay
+ * rebuilds the identical graph. Soft lock: later high-priority steering can still preempt.
+ */
+@Serializable
+@SerialName("ExecutionPlanLocked")
+data class ExecutionPlanLockedEvent(
+    val sessionId: SessionId,
+    val planArtifactId: ArtifactId,
+    val workflowId: String,
+    val stageIds: List<String>,
+    val startStageId: String,
+) : EventPayload
@@ -13,6 +13,7 @@ import com.correx.core.events.events.RouterNarrationEvent
 import com.correx.core.events.events.SessionWorkspaceBoundEvent
 import com.correx.core.events.events.ContextTruncatedEvent
 import com.correx.core.events.events.EventPayload
+import com.correx.core.events.events.ExecutionPlanLockedEvent
 import com.correx.core.events.events.FileWrittenEvent
 import com.correx.core.events.events.L3MemoryRetrievedEvent
 import com.correx.core.events.events.InferenceCompletedEvent
@@ -91,6 +92,7 @@ val eventModule = SerializersModule {
        subclass(SessionWorkspaceBoundEvent::class)
        subclass(L3MemoryRetrievedEvent::class)
        subclass(ContextTruncatedEvent::class)
+        subclass(ExecutionPlanLockedEvent::class)
    }
 }

@@ -0,0 +1,27 @@
+package com.correx.core.events.serialization
+
+import com.correx.core.events.events.EventPayload
+import com.correx.core.events.events.ExecutionPlanLockedEvent
+import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.types.SessionId
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class ExecutionPlanLockedEventSerializationTest {
+
+    @Test
+    fun `ExecutionPlanLockedEvent round-trips as polymorphic EventPayload`() {
+        val sample: EventPayload = ExecutionPlanLockedEvent(
+            sessionId = SessionId("s"),
+            planArtifactId = ArtifactId("execution_plan"),
+            workflowId = "freestyle-s",
+            stageIds = listOf("impl", "review"),
+            startStageId = "impl",
+        )
+        val encoded = eventJson.encodeToString(EventPayload.serializer(), sample)
+        assertTrue(encoded.contains("\"type\":\"ExecutionPlanLocked\""), "SerialName must be present: $encoded")
+        val decoded = eventJson.decodeFromString(EventPayload.serializer(), encoded)
+        assertEquals(sample, decoded)
+    }
+}
@@ -0,0 +1,68 @@
+package com.correx.infrastructure.workflow
+
+import com.correx.core.artifacts.kind.ArtifactKindRegistry
+import com.correx.core.artifacts.kind.TypedArtifactSlot
+import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.types.StageId
+import com.correx.core.events.types.TransitionId
+import com.correx.core.transitions.graph.StageConfig
+import com.correx.core.transitions.graph.TransitionEdge
+import com.correx.core.transitions.graph.WorkflowGraph
+import com.fasterxml.jackson.databind.DeserializationFeature
+import com.fasterxml.jackson.databind.json.JsonMapper
+import com.fasterxml.jackson.module.kotlin.kotlinModule
+import com.fasterxml.jackson.module.kotlin.readValue
+
+private const val TERMINAL = "done"
+
+class ExecutionPlanCompiler(
+    private val registry: ArtifactKindRegistry,
+) {
+    private val mapper = JsonMapper.builder()
+        .addModule(kotlinModule())
+        .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+        .build()
+
+    fun compile(planJson: String, workflowId: String): WorkflowGraph {
+        val plan = runCatching { mapper.readValue<ExecutionPlanModel>(planJson) }
+            .getOrElse { throw WorkflowValidationException("Failed to parse execution_plan JSON: ${it.message}") }
+        if (plan.stages.isEmpty()) throw WorkflowValidationException("execution_plan has no stages")
+
+        val stageMap = plan.stages.associate { s ->
+            val kind = registry.get(s.produces)
+                ?: throw WorkflowValidationException("Unknown artifact kind '${s.produces}' in stage '${s.id}'")
+            StageId(s.id) to StageConfig(
+                produces = listOf(TypedArtifactSlot(name = ArtifactId(s.produces), kind = kind)),
+                needs = s.needs.map { ArtifactId(it) }.toSet(),
+                allowedTools = s.tools.toSet(),
+                metadata = mapOf("promptInline" to s.prompt),
+            )
+        }
+        val declared = stageMap.keys.map { it.value }.toSet()
+
+        val transitions = plan.edges.map { e ->
+            if (e.from !in declared) {
+                throw WorkflowValidationException("edge from unknown stage '${e.from}'")
+            }
+            if (e.to !in declared && e.to != TERMINAL) {
+                throw WorkflowValidationException("edge to unknown stage '${e.to}'")
+            }
+            TransitionEdge(
+                id = TransitionId("${e.from}->${e.to}"),
+                from = StageId(e.from),
+                to = StageId(e.to),
+                condition = ConditionSpec(
+                    type = e.condition.type,
+                    artifactId = e.condition.artifactId,
+                    key = e.condition.key,
+                    field = e.condition.field,
+                    value = e.condition.value,
+                    operator = e.condition.operator,
+                ).toCondition(),
+            )
+        }.toSet()
+
+        val start = StageId(plan.stages.first().id)
+        return WorkflowGraph(id = workflowId, stages = stageMap, transitions = transitions, start = start)
+    }
+}
@@ -0,0 +1,32 @@
+package com.correx.infrastructure.workflow
+
+import com.fasterxml.jackson.annotation.JsonProperty
+
+data class ExecutionPlanModel(
+    val goal: String = "",
+    val stages: List<PlanStage> = emptyList(),
+    val edges: List<PlanEdge> = emptyList(),
+)
+
+data class PlanStage(
+    val id: String = "",
+    val prompt: String = "",
+    val produces: String = "",
+    val needs: List<String> = emptyList(),
+    val tools: List<String> = emptyList(),
+)
+
+data class PlanEdge(
+    val from: String = "",
+    val to: String = "",
+    val condition: PlanCondition = PlanCondition(),
+)
+
+data class PlanCondition(
+    val type: String = "always_true",
+    @param:JsonProperty("artifact_id") val artifactId: String? = null,
+    val key: String? = null,
+    val field: String? = null,
+    val value: String? = null,
+    val operator: String? = null,
+)
@@ -0,0 +1,100 @@
+package com.correx.infrastructure.workflow
+
+import com.correx.core.artifacts.kind.ConfigArtifactKind
+import com.correx.core.artifacts.kind.DefaultArtifactKindRegistry
+import com.correx.core.artifacts.kind.JsonSchema
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+class ExecutionPlanCompilerTest {
+
+    private val trivialSchema = JsonSchema(type = "object")
+    private val registry = DefaultArtifactKindRegistry().also {
+        it.register(ConfigArtifactKind(id = "patch", schema = trivialSchema, llmEmitted = true))
+    }
+    private val compiler = ExecutionPlanCompiler(registry)
+
+    private val validPlan = """
+        {
+          "goal": "apply a patch",
+          "stages": [
+            {
+              "id": "analyse",
+              "prompt": "Analyse the codebase",
+              "produces": "patch",
+              "needs": [],
+              "tools": ["ShellTool"]
+            },
+            {
+              "id": "review",
+              "prompt": "Review the patch",
+              "produces": "patch",
+              "needs": ["patch"],
+              "tools": []
+            }
+          ],
+          "edges": [
+            {
+              "from": "analyse",
+              "to": "review",
+              "condition": { "type": "always_true" }
+            },
+            {
+              "from": "review",
+              "to": "done",
+              "condition": { "type": "always_true" }
+            }
+          ]
+        }
+    """.trimIndent()
+
+    @Test
+    fun `valid plan produces correct WorkflowGraph`() {
+        val graph = compiler.compile(validPlan, "test-workflow")
+
+        assertEquals("test-workflow", graph.id)
+        assertEquals(2, graph.stages.size)
+        assertEquals(2, graph.transitions.size)
+        assertEquals("analyse", graph.start.value)
+
+        val analyseStage = graph.stages[graph.start]
+        assertNotNull(analyseStage)
+        assertEquals("Analyse the codebase", analyseStage.metadata["promptInline"])
+        assertEquals(setOf("patch"), analyseStage.produces.map { it.name.value }.toSet())
+        assertEquals(setOf("ShellTool"), analyseStage.allowedTools)
+
+        val reviewStage = graph.stages.values.first { it.metadata["promptInline"] == "Review the patch" }
+        assertEquals(setOf("patch"), reviewStage.needs.map { it.value }.toSet())
+    }
+
+    @Test
+    fun `edge referencing unknown from-stage throws WorkflowValidationException`() {
+        val bad = validPlan.replace("\"from\": \"analyse\"", "\"from\": \"nonexistent\"")
+        assertThrows<WorkflowValidationException> { compiler.compile(bad, "bad-workflow") }
+    }
+
+    @Test
+    fun `edge referencing unknown to-stage throws WorkflowValidationException`() {
+        val bad = validPlan.replace("\"to\": \"review\"", "\"to\": \"nonexistent\"")
+        assertThrows<WorkflowValidationException> { compiler.compile(bad, "bad-workflow") }
+    }
+
+    @Test
+    fun `unregistered produces kind throws WorkflowValidationException`() {
+        val bad = validPlan.replace("\"produces\": \"patch\"", "\"produces\": \"unknown_kind\"")
+        assertThrows<WorkflowValidationException> { compiler.compile(bad, "bad-workflow") }
+    }
+
+    @Test
+    fun `empty stages throws WorkflowValidationException`() {
+        val bad = """{ "goal": "x", "stages": [], "edges": [] }"""
+        assertThrows<WorkflowValidationException> { compiler.compile(bad, "bad-workflow") }
+    }
+
+    @Test
+    fun `malformed JSON throws WorkflowValidationException`() {
+        assertThrows<WorkflowValidationException> { compiler.compile("not-json", "bad-workflow") }
+    }
+}
@@ -14,4 +14,6 @@ dependencies {
    testImplementation(project(":core:approvals"))
    testImplementation(project(":testing:fixtures"))
    testImplementation(project(":infrastructure:persistence"))
+    testImplementation(project(":infrastructure:workflow"))
+    testImplementation(project(":core:artifacts"))
 }
@@ -0,0 +1,103 @@
+import com.correx.core.artifacts.kind.ConfigArtifactKind
+import com.correx.core.artifacts.kind.DefaultArtifactKindRegistry
+import com.correx.core.artifacts.kind.JsonSchema
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.ExecutionPlanLockedEvent
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.types.EventId
+import com.correx.core.events.types.SessionId
+import com.correx.infrastructure.persistence.InMemoryEventStore
+import com.correx.infrastructure.workflow.ExecutionPlanCompiler
+import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Clock
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+
+class ExecutionPlanLockedReplayTest {
+
+    private val planJson = """
+        {
+          "goal": "replay test plan",
+          "stages": [
+            {
+              "id": "fetch",
+              "prompt": "Fetch the data",
+              "produces": "patch",
+              "needs": [],
+              "tools": []
+            },
+            {
+              "id": "transform",
+              "prompt": "Transform the data",
+              "produces": "patch",
+              "needs": ["patch"],
+              "tools": []
+            }
+          ],
+          "edges": [
+            {
+              "from": "fetch",
+              "to": "transform",
+              "condition": { "type": "always_true" }
+            },
+            {
+              "from": "transform",
+              "to": "done",
+              "condition": { "type": "always_true" }
+            }
+          ]
+        }
+    """.trimIndent()
+
+    @Test
+    fun `locked plan event recompiles to identical WorkflowGraph`(): Unit = runBlocking {
+        val sessionId = SessionId("replay-session-1")
+        val planArtifactId = ArtifactId("cas://plan-artifact-001")
+
+        // CAS store: keyed by ArtifactId -> plan JSON (no FS access — replay determinism)
+        val casStore = mutableMapOf<ArtifactId, String>()
+        casStore[planArtifactId] = planJson
+
+        val lockedEvent = ExecutionPlanLockedEvent(
+            sessionId = sessionId,
+            planArtifactId = planArtifactId,
+            workflowId = "replay-workflow",
+            stageIds = listOf("fetch", "transform"),
+            startStageId = "fetch",
+        )
+
+        val store = InMemoryEventStore()
+        store.append(
+            NewEvent(
+                metadata = EventMetadata(
+                    eventId = EventId("locked-evt-1"),
+                    sessionId = sessionId,
+                    timestamp = Clock.System.now(),
+                    schemaVersion = 1,
+                    causationId = null,
+                    correlationId = null,
+                ),
+                payload = lockedEvent,
+            )
+        )
+
+        // Replay: read event back, fetch JSON from CAS, recompile — no FS, no LLM
+        val replayedEvent = store.read(sessionId)
+            .map { it.payload }
+            .filterIsInstance<ExecutionPlanLockedEvent>()
+            .single()
+
+        val replayedJson = casStore.getValue(replayedEvent.planArtifactId)
+
+        val trivialSchema = JsonSchema(type = "object")
+        val registry = DefaultArtifactKindRegistry().also {
+            it.register(ConfigArtifactKind(id = "patch", schema = trivialSchema, llmEmitted = true))
+        }
+        val compiler = ExecutionPlanCompiler(registry)
+        val graph = compiler.compile(replayedJson, replayedEvent.workflowId)
+
+        assertEquals(replayedEvent.stageIds.toSet(), graph.stages.keys.map { it.value }.toSet())
+        assertEquals(replayedEvent.startStageId, graph.start.value)
+    }
+}