kami
6c8c5e2ad9
- 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.
104 lines
3.6 KiB
Kotlin
104 lines
3.6 KiB
Kotlin
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)
|
|
}
|
|
}
|