From 6c8c5e2ad9484acdf0b585a1d84873b621443868 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 8 Jun 2026 02:33:57 +0400 Subject: [PATCH] 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. --- .../core/events/events/ExecutionPlanEvents.kt | 22 ++++ .../events/serialization/Serialization.kt | 2 + ...ecutionPlanLockedEventSerializationTest.kt | 27 +++++ .../workflow/ExecutionPlanCompiler.kt | 68 ++++++++++++ .../workflow/ExecutionPlanModel.kt | 32 ++++++ .../workflow/ExecutionPlanCompilerTest.kt | 100 +++++++++++++++++ testing/replay/build.gradle | 2 + .../kotlin/ExecutionPlanLockedReplayTest.kt | 103 ++++++++++++++++++ 8 files changed, 356 insertions(+) create mode 100644 core/events/src/main/kotlin/com/correx/core/events/events/ExecutionPlanEvents.kt create mode 100644 core/events/src/test/kotlin/com/correx/core/events/serialization/ExecutionPlanLockedEventSerializationTest.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt create mode 100644 infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt create mode 100644 testing/replay/src/test/kotlin/ExecutionPlanLockedReplayTest.kt diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/ExecutionPlanEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/ExecutionPlanEvents.kt new file mode 100644 index 00000000..59a2c7f6 --- /dev/null +++ b/core/events/src/main/kotlin/com/correx/core/events/events/ExecutionPlanEvents.kt @@ -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, + val startStageId: String, +) : 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 6b5b4b84..bf5efde9 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 @@ -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) } } diff --git a/core/events/src/test/kotlin/com/correx/core/events/serialization/ExecutionPlanLockedEventSerializationTest.kt b/core/events/src/test/kotlin/com/correx/core/events/serialization/ExecutionPlanLockedEventSerializationTest.kt new file mode 100644 index 00000000..f83e8b90 --- /dev/null +++ b/core/events/src/test/kotlin/com/correx/core/events/serialization/ExecutionPlanLockedEventSerializationTest.kt @@ -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) + } +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt new file mode 100644 index 00000000..8191237e --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompiler.kt @@ -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(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) + } +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt new file mode 100644 index 00000000..b676b1c6 --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ExecutionPlanModel.kt @@ -0,0 +1,32 @@ +package com.correx.infrastructure.workflow + +import com.fasterxml.jackson.annotation.JsonProperty + +data class ExecutionPlanModel( + val goal: String = "", + val stages: List = emptyList(), + val edges: List = emptyList(), +) + +data class PlanStage( + val id: String = "", + val prompt: String = "", + val produces: String = "", + val needs: List = emptyList(), + val tools: List = 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, +) diff --git a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt new file mode 100644 index 00000000..869cb912 --- /dev/null +++ b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/ExecutionPlanCompilerTest.kt @@ -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 { compiler.compile(bad, "bad-workflow") } + } + + @Test + fun `edge referencing unknown to-stage throws WorkflowValidationException`() { + val bad = validPlan.replace("\"to\": \"review\"", "\"to\": \"nonexistent\"") + assertThrows { compiler.compile(bad, "bad-workflow") } + } + + @Test + fun `unregistered produces kind throws WorkflowValidationException`() { + val bad = validPlan.replace("\"produces\": \"patch\"", "\"produces\": \"unknown_kind\"") + assertThrows { compiler.compile(bad, "bad-workflow") } + } + + @Test + fun `empty stages throws WorkflowValidationException`() { + val bad = """{ "goal": "x", "stages": [], "edges": [] }""" + assertThrows { compiler.compile(bad, "bad-workflow") } + } + + @Test + fun `malformed JSON throws WorkflowValidationException`() { + assertThrows { compiler.compile("not-json", "bad-workflow") } + } +} diff --git a/testing/replay/build.gradle b/testing/replay/build.gradle index 10fe62b0..d5e1a17d 100644 --- a/testing/replay/build.gradle +++ b/testing/replay/build.gradle @@ -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")) } \ No newline at end of file diff --git a/testing/replay/src/test/kotlin/ExecutionPlanLockedReplayTest.kt b/testing/replay/src/test/kotlin/ExecutionPlanLockedReplayTest.kt new file mode 100644 index 00000000..a90fe493 --- /dev/null +++ b/testing/replay/src/test/kotlin/ExecutionPlanLockedReplayTest.kt @@ -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() + 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() + .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) + } +}