diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 8941dc8e..4a41ade2 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -54,8 +54,12 @@ import com.correx.core.toolintent.WorkspacePolicy import com.correx.core.toolintent.rules.ExecInterpreterRule import com.correx.core.toolintent.rules.NetworkHostRule import com.correx.core.toolintent.rules.PathContainmentRule +import com.correx.apps.server.freestyle.FreestyleDriver +import com.correx.core.artifacts.kind.DefaultArtifactKindRegistry +import com.correx.core.events.types.ArtifactId import com.correx.infrastructure.InfrastructureModule import com.correx.infrastructure.inference.DefaultProviderRegistry +import com.correx.infrastructure.workflow.ExecutionPlanCompiler import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy import com.correx.infrastructure.inference.commons.ManagedInferenceRouter import com.correx.infrastructure.inference.commons.AmdResourceProbe @@ -306,13 +310,24 @@ fun main() { } else { null } + val configArtifactKinds = loadConfigArtifactKinds(correxConfig) + val artifactKindRegistry = DefaultArtifactKindRegistry().also { reg -> + configArtifactKinds.forEach { reg.register(it) } + } + val freestyleDriver = FreestyleDriver( + eventStore = eventStore, + compiler = ExecutionPlanCompiler(artifactKindRegistry), + planContent = { sid -> orchestrator.validatedArtifactContent(sid, ArtifactId("execution_plan")) }, + config = defaultOrchestrationConfig, + runPhase2 = { sid, graph, cfg -> orchestrator.run(sid, graph, cfg) }, + ) val module = ServerModule( orchestrator = orchestrator, eventStore = eventStore, artifactStore = artifactStore, sessionRepository = repositories.sessionRepository, workflowRegistry = FileSystemWorkflowRegistry( - InfrastructureModule.createWorkflowLoader(loadConfigArtifactKinds(correxConfig)), + InfrastructureModule.createWorkflowLoader(configArtifactKinds), ), providerRegistry = infraRegistry.asServerRegistry(), defaultOrchestrationConfig = defaultOrchestrationConfig, @@ -326,6 +341,7 @@ fun main() { workspaceResolver = workspaceResolver, narrationMaxPerRun = routerConfig.narration.maxPerRun, projectMemory = projectMemory, + freestyleDriver = freestyleDriver, ) module.start() log.info("==============================") diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index d9a36557..82413d10 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -1,6 +1,7 @@ package com.correx.apps.server import com.correx.apps.server.approval.ApprovalCoordinator +import com.correx.apps.server.freestyle.FreestyleDriver import com.correx.apps.server.narration.NarrationSubscriber import com.correx.apps.server.registry.ProviderRegistry import com.correx.apps.server.registry.WorkflowRegistry @@ -69,6 +70,9 @@ class ServerModule( val narrationMaxPerRun: Int = 100, // Cross-session, repo-scoped memory. Null disables the hooks entirely (tests / static path). val projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = null, + // Driver for the two-phase freestyle workflow: locks the plan and runs phase 2. + // Null on non-freestyle paths; injected by Main when freestyle is configured. + private val freestyleDriver: FreestyleDriver? = null, ) { val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator( orchestrator = orchestrator, @@ -139,6 +143,10 @@ class ServerModule( } runCatching { orchestrator.run(sessionId, graph, sessionConfig) + // After a successful freestyle planning run, lock the plan and execute phase 2. + if (graph.id == "freestyle_planning") { + freestyleDriver?.lockAndRun(sessionId) + } // Distil this run's decisions into durable project memory on completion. projectMemory?.let { pm -> pm.persist(sessionId, pm.repoRoot()) } }.onFailure { ex -> diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/freestyle/FreestyleDriver.kt b/apps/server/src/main/kotlin/com/correx/apps/server/freestyle/FreestyleDriver.kt new file mode 100644 index 00000000..4335d790 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/freestyle/FreestyleDriver.kt @@ -0,0 +1,67 @@ +package com.correx.apps.server.freestyle + +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.stores.EventStore +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.kernel.execution.WorkflowResult +import com.correx.core.kernel.orchestration.OrchestrationConfig +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.infrastructure.workflow.ExecutionPlanCompiler +import kotlinx.datetime.Clock +import org.slf4j.LoggerFactory +import java.util.UUID + +/** + * Drives the second phase of a freestyle workflow: reads the validated execution_plan, + * compiles it, emits [ExecutionPlanLockedEvent], and invokes phase-2 orchestration. + * + * [runPhase2] is a functional seam so the driver can be tested without the full + * [DefaultSessionOrchestrator] constructor chain. + */ +class FreestyleDriver( + private val eventStore: EventStore, + private val compiler: ExecutionPlanCompiler, + private val planContent: (SessionId) -> String?, + private val config: OrchestrationConfig, + private val runPhase2: suspend (SessionId, WorkflowGraph, OrchestrationConfig) -> WorkflowResult, +) { + suspend fun lockAndRun(sessionId: SessionId) { + val json = planContent(sessionId) ?: run { + log.warn("freestyle: no execution_plan content for session={}", sessionId.value) + return + } + val graph = runCatching { compiler.compile(json, "freestyle-${sessionId.value}") } + .getOrElse { + log.error("freestyle: plan failed to compile: {}", it.message) + return + } + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = ExecutionPlanLockedEvent( + sessionId = sessionId, + planArtifactId = ArtifactId("execution_plan"), + workflowId = graph.id, + stageIds = graph.stageIds.map { it.value }, + startStageId = graph.start.value, + ), + ), + ) + runPhase2(sessionId, graph, config) + } + + companion object { + private val log = LoggerFactory.getLogger(FreestyleDriver::class.java) + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/freestyle/FreestyleDriverTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/freestyle/FreestyleDriverTest.kt new file mode 100644 index 00000000..d88b47f3 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/freestyle/FreestyleDriverTest.kt @@ -0,0 +1,157 @@ +package com.correx.apps.server.freestyle + +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.ExecutionPlanLockedEvent +import com.correx.core.events.types.SessionId +import com.correx.core.kernel.execution.WorkflowResult +import com.correx.core.kernel.orchestration.OrchestrationConfig +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.infrastructure.workflow.ExecutionPlanCompiler +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class FreestyleDriverTest { + + private val validPlanJson = """ + { + "goal": "test plan", + "stages": [ + { + "id": "analyse", + "prompt": "Analyse the problem", + "produces": "patch", + "needs": [], + "tools": [] + }, + { + "id": "apply", + "prompt": "Apply the fix", + "produces": "patch", + "needs": ["patch"], + "tools": [] + } + ], + "edges": [ + { + "from": "analyse", + "to": "apply", + "condition": { "type": "always_true" } + }, + { + "from": "apply", + "to": "done", + "condition": { "type": "always_true" } + } + ] + } + """.trimIndent() + + private val malformedPlanJson = """{ "not": "a valid plan" }""" + + private fun buildRegistry(): DefaultArtifactKindRegistry = + DefaultArtifactKindRegistry().also { + it.register(ConfigArtifactKind(id = "patch", schema = JsonSchema(type = "object"), llmEmitted = true)) + } + + @Test + fun `lockAndRun emits ExecutionPlanLockedEvent with compiled stageIds and invokes runPhase2`(): Unit = runBlocking { + val sessionId = SessionId("driver-test-session") + val eventStore = InMemoryEventStore() + val registry = buildRegistry() + val compiler = ExecutionPlanCompiler(registry) + + var runPhase2Invocations = 0 + var capturedGraph: WorkflowGraph? = null + + val driver = FreestyleDriver( + eventStore = eventStore, + compiler = compiler, + planContent = { validPlanJson }, + config = OrchestrationConfig(), + runPhase2 = { sid, graph, _ -> + runPhase2Invocations++ + capturedGraph = graph + WorkflowResult.Completed(sid, graph.start) + }, + ) + + driver.lockAndRun(sessionId) + + val lockedEvents = eventStore.read(sessionId) + .map { it.payload } + .filterIsInstance() + + assertEquals(1, lockedEvents.size, "Expected exactly one ExecutionPlanLockedEvent") + val locked = lockedEvents.single() + assertEquals(setOf("analyse", "apply"), locked.stageIds.toSet()) + assertEquals("analyse", locked.startStageId) + assertTrue(locked.workflowId.startsWith("freestyle-")) + assertEquals(1, runPhase2Invocations, "runPhase2 should be called exactly once") + assertEquals(setOf("analyse", "apply"), capturedGraph?.stageIds?.map { it.value }?.toSet()) + } + + @Test + fun `lockAndRun with malformed plan emits no lock event and does not invoke runPhase2`(): Unit = runBlocking { + val sessionId = SessionId("driver-malformed-session") + val eventStore = InMemoryEventStore() + val registry = buildRegistry() + val compiler = ExecutionPlanCompiler(registry) + + var runPhase2Invocations = 0 + + val driver = FreestyleDriver( + eventStore = eventStore, + compiler = compiler, + planContent = { malformedPlanJson }, + config = OrchestrationConfig(), + runPhase2 = { _, _, _ -> + runPhase2Invocations++ + WorkflowResult.Completed(sessionId, com.correx.core.events.types.StageId("done")) + }, + ) + + driver.lockAndRun(sessionId) + + val lockedEvents = eventStore.read(sessionId) + .map { it.payload } + .filterIsInstance() + + assertTrue(lockedEvents.isEmpty(), "No lock event expected for malformed plan") + assertEquals(0, runPhase2Invocations, "runPhase2 must not be called for malformed plan") + } + + @Test + fun `lockAndRun with absent plan content does not emit lock event or invoke runPhase2`(): Unit = runBlocking { + val sessionId = SessionId("driver-absent-session") + val eventStore = InMemoryEventStore() + val registry = buildRegistry() + val compiler = ExecutionPlanCompiler(registry) + + var runPhase2Invocations = 0 + + val driver = FreestyleDriver( + eventStore = eventStore, + compiler = compiler, + planContent = { null }, + config = OrchestrationConfig(), + runPhase2 = { _, _, _ -> + runPhase2Invocations++ + WorkflowResult.Completed(sessionId, com.correx.core.events.types.StageId("done")) + }, + ) + + driver.lockAndRun(sessionId) + + val lockedEvents = eventStore.read(sessionId) + .map { it.payload } + .filterIsInstance() + + assertTrue(lockedEvents.isEmpty(), "No lock event when plan content is absent") + assertEquals(0, runPhase2Invocations, "runPhase2 must not be called when plan is absent") + } +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 50a6c56e..1290600c 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -8,6 +8,7 @@ import com.correx.core.journal.DefaultDecisionJournalRepository import com.correx.core.artifacts.ArtifactState import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RefinementIterationEvent @@ -155,6 +156,10 @@ class DefaultSessionOrchestrator( } + /** Returns the cached validated artifact content for the given session + artifact id, or null if absent. */ + fun validatedArtifactContent(sessionId: SessionId, artifactId: ArtifactId): String? = + artifactContentCache["${sessionId.value}:${artifactId.value}"] + suspend fun resume( sessionId: SessionId, graph: WorkflowGraph, @@ -254,6 +259,33 @@ class DefaultSessionOrchestrator( stageId: StageId, ): StepResult { log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}") + + if (ctx.graph.stages[stageId]?.metadata?.get("requiresApproval") == "true") { + val alreadyApproved = repositories.eventStore.read(ctx.sessionId).let { events -> + val stageRequestIds = events + .mapNotNull { it.payload as? ApprovalRequestedEvent } + .filter { it.stageId == stageId && it.toolName == null } + .map { it.requestId } + .toSet() + stageRequestIds.isNotEmpty() && events.any { + (it.payload as? ApprovalDecisionResolvedEvent) + ?.requestId in stageRequestIds + } + } + if (!alreadyApproved) { + val gateArtifact = ctx.graph.stages[stageId]?.needs?.firstOrNull()?.value + val preview = gateArtifact + ?.let { artifactContentCache["${ctx.sessionId.value}:$it"] } + ?: "Upstream stage has completed. Review and approve to continue to stage ${stageId.value}." + val approved = requestStageApproval(ctx.sessionId, stageId, preview) + if (!approved) { + return StepResult.Terminal( + failWorkflow(ctx.sessionId, stageId, "approval rejected for stage ${stageId.value}", retryExhausted = false) + ) + } + } + } + while (true) { if (isCancelled(ctx.sessionId)) { return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId)) 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 feb35dca..a55ce511 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 @@ -250,21 +250,9 @@ abstract class SessionOrchestrator( ), ) } ?: emptyList() - val promptEntries = stageConfig.metadata["prompt"] - ?.let { path -> - val resolvedText = runCatching { promptResolver.resolve(path) } - .onFailure { - log.error( - "[SessionOrchestrator] stage=${stageId.value}: " + - "failed to load prompt '$path': ${it.message}", - ) - } - .getOrNull() - val text = resolvedText?.takeIf { it.isNotBlank() } - ?: error( - "[SessionOrchestrator] stage=${stageId.value}: " + - "declared prompt '$path' could not be resolved", - ) + val promptEntries = stageConfig.metadata["promptInline"] + ?.takeIf { it.isNotBlank() } + ?.let { text -> listOf( ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), @@ -276,7 +264,35 @@ abstract class SessionOrchestrator( role = EntryRole.USER, ), ) - } ?: emptyList() + } + ?: stageConfig.metadata["prompt"] + ?.let { path -> + val resolvedText = runCatching { promptResolver.resolve(path) } + .onFailure { + log.error( + "[SessionOrchestrator] stage=${stageId.value}: " + + "failed to load prompt '$path': ${it.message}", + ) + } + .getOrNull() + val text = resolvedText?.takeIf { it.isNotBlank() } + ?: error( + "[SessionOrchestrator] stage=${stageId.value}: " + + "declared prompt '$path' could not be resolved", + ) + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L1, + content = text, + sourceType = "agentPrompt", + sourceId = stageId.value, + tokenEstimate = estimateTokens(text), + role = EntryRole.USER, + ), + ) + } + ?: emptyList() val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted } require(llmEmittedSlots.size <= 1) { @@ -1205,6 +1221,65 @@ abstract class SessionOrchestrator( return (content.length / 4).coerceAtLeast(1) } + // --- stage-entry approval gate --- + + /** + * Emits [OrchestrationPausedEvent] + [ApprovalRequestedEvent] for a stage flagged + * `requiresApproval`, awaits the user decision, and records it. + * Returns `true` if the run should proceed into the stage, `false` if the approval was rejected. + */ + internal suspend fun requestStageApproval( + sessionId: SessionId, + stageId: StageId, + preview: String, + ): Boolean { + val requestId = ApprovalRequestId(UUID.randomUUID().toString()) + val validationReportId = ValidationReportId(UUID.randomUUID().toString()) + val deferred = CompletableDeferred() + pendingApprovals[requestId] = deferred + + emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING")) + emit( + sessionId, + ApprovalRequestedEvent( + requestId = requestId, + tier = Tier.T2, + validationReportId = validationReportId, + riskSummaryId = null, + riskSummary = null, + sessionId = sessionId, + stageId = stageId, + projectId = null, + preview = preview, + ), + ) + + return try { + val decision = deferred.await() + emitDecisionResolved( + sessionId, + DomainApprovalRequest( + id = requestId, + tier = Tier.T2, + validationReportId = validationReportId, + riskSummaryId = null, + riskSummary = null, + timestamp = Clock.System.now(), + ), + decision, + ) + if (decision.isApproved) { + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) + true + } else { + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) + false + } + } finally { + pendingApprovals.remove(requestId) + } + } + // --- private functions --- private suspend fun handleApproval( diff --git a/examples/workflows/freestyle_planning.toml b/examples/workflows/freestyle_planning.toml new file mode 100644 index 00000000..5154644c --- /dev/null +++ b/examples/workflows/freestyle_planning.toml @@ -0,0 +1,33 @@ +id = "freestyle_planning" +start = "analyst" + +[[stages]] +id = "analyst" +prompt = "prompts/analyst_freestyle.md" +produces = [{ name = "analysis", kind = "analysis" }] +allowed_tools = ["file_read", "ShellTool"] +token_budget = 16384 +max_retries = 2 + +[[stages]] +id = "architect" +requires_approval = true +prompt = "prompts/architect_freestyle.md" +needs = ["analysis"] +produces = [{ name = "execution_plan", kind = "execution_plan" }] +token_budget = 16384 +max_retries = 2 + +[[transitions]] +id = "analyst-to-architect" +from = "analyst" +to = "architect" +condition_type = "artifact_validated" +condition_artifact_id = "analysis" + +[[transitions]] +id = "architect-to-done" +from = "architect" +to = "done" +condition_type = "artifact_validated" +condition_artifact_id = "execution_plan" diff --git a/examples/workflows/prompts/analyst_freestyle.md b/examples/workflows/prompts/analyst_freestyle.md new file mode 100644 index 00000000..b12b1014 --- /dev/null +++ b/examples/workflows/prompts/analyst_freestyle.md @@ -0,0 +1,10 @@ +You are the **Analyst** in freestyle mode. Understand the user's goal (in the decision history +above) and the code it touches. Read-only: `file_read`, `ls`, `grep`, `cat`, `find`. + +Emit the `analysis` artifact (JSON, schema provided): +- `summary`: the goal in your own words, AND any open questions the user must answer before a + plan can be built — prefix each question with "Q:". If none, say "No open questions." +- `requirements`: concrete, checkable requirements, one per line. +- `affected_areas`: files/modules likely involved, one per line. + +Do not design or plan yet. diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt index 6f7a96a2..afad9464 100644 --- a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt @@ -39,6 +39,7 @@ private data class StageSection( @param:JsonProperty("allowed_tools") val allowedTools: List = emptyList(), @param:JsonProperty("token_budget") val tokenBudget: Int = 4096, @param:JsonProperty("max_retries") val maxRetries: Int = 3, + @param:JsonProperty("requires_approval") val requiresApproval: Boolean = false, ) // condition fields flattened into the transition row @@ -94,6 +95,7 @@ class TomlWorkflowLoader( metadata = buildMap { s.prompt?.let { put("prompt", resolvePromptPath(it, workflowDir)) } s.systemPrompt?.let { put("systemPrompt", resolvePromptPath(it, workflowDir)) } + if (s.requiresApproval) put("requiresApproval", "true") }, ) } diff --git a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FreestylePlanningWorkflowTest.kt b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FreestylePlanningWorkflowTest.kt new file mode 100644 index 00000000..bd238835 --- /dev/null +++ b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FreestylePlanningWorkflowTest.kt @@ -0,0 +1,73 @@ +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 com.correx.core.artifacts.kind.JsonSchemaProperty +import com.correx.core.events.types.StageId +import com.correx.core.transitions.conditions.ArtifactValidated +import org.junit.jupiter.api.Assumptions.assumeTrue +import org.junit.jupiter.api.Test +import java.nio.file.Path +import kotlin.io.path.exists +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +class FreestylePlanningWorkflowTest { + + private fun llmKind(id: String) = ConfigArtifactKind( + id = id, + schema = JsonSchema( + type = "object", + properties = mapOf("summary" to JsonSchemaProperty(type = "string")), + additionalProperties = true, + ), + llmEmitted = true, + ) + + private val registry = DefaultArtifactKindRegistry().apply { + register(llmKind("analysis")) + register(llmKind("execution_plan")) + } + + private fun repoFile(rel: String): Path? { + var dir: Path? = Path.of(System.getProperty("user.dir")).toAbsolutePath() + while (dir != null) { + val candidate = dir.resolve(rel) + if (candidate.exists()) return candidate + dir = dir.parent + } + return null + } + + @Test + fun `freestyle_planning toml parses with two non-terminal stages and both edges`() { + val path = repoFile("examples/workflows/freestyle_planning.toml") + assumeTrue(path != null, "freestyle_planning.toml not found from ${System.getProperty("user.dir")}") + + val graph = TomlWorkflowLoader(registry).load(path!!) + + assertEquals( + setOf("analyst", "architect"), + graph.stages.keys.map { it.value }.toSet(), + ) + + val architect = graph.stages[StageId("architect")]!! + assertTrue( + architect.produces.any { it.name.value == "execution_plan" && it.kind.llmEmitted }, + "architect must produce execution_plan", + ) + + assertTrue(architect.needs.map { it.value }.contains("analysis"), "architect needs analysis") + + val analystToArchitect = graph.transitions.first { it.from == StageId("analyst") } + assertEquals(StageId("architect"), analystToArchitect.to) + assertTrue(analystToArchitect.condition is ArtifactValidated) + + val architectToDone = graph.transitions.first { it.from == StageId("architect") } + assertEquals(StageId("done"), architectToDone.to) + assertTrue(architectToDone.condition is ArtifactValidated) + + assertEquals(2, graph.transitions.size) + } +} diff --git a/testing/integration/src/test/kotlin/FreestyleApprovalGateTest.kt b/testing/integration/src/test/kotlin/FreestyleApprovalGateTest.kt new file mode 100644 index 00000000..76d3d7f1 --- /dev/null +++ b/testing/integration/src/test/kotlin/FreestyleApprovalGateTest.kt @@ -0,0 +1,254 @@ +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalProjector +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.DefaultApprovalReducer +import com.correx.core.approvals.DefaultApprovalRepository +import com.correx.core.approvals.Tier +import com.correx.core.approvals.domain.DefaultApprovalEngine +import com.correx.core.approvals.model.ApprovalContext +import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.approvals.model.ApprovalScopeIdentity +import com.correx.core.artifacts.DefaultArtifactReducer +import com.correx.core.artifacts.kind.ConfigArtifactKind +import com.correx.core.artifacts.kind.JsonSchema +import com.correx.core.artifacts.kind.TypedArtifactSlot +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.OrchestrationPausedEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.execution.RetryPolicy +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.TransitionId +import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.InferenceState +import com.correx.core.inference.ModelCapability +import com.correx.core.journal.DecisionJournalProjector +import com.correx.core.journal.DefaultDecisionJournalReducer +import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer +import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.kernel.orchestration.OrchestrationConfig +import com.correx.core.kernel.orchestration.OrchestrationProjector +import com.correx.core.kernel.orchestration.OrchestrationRepository +import com.correx.core.kernel.orchestration.OrchestratorEngines +import com.correx.core.kernel.orchestration.OrchestratorRepositories +import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.risk.DefaultRiskAssessor +import com.correx.core.sessions.ApprovalMode +import com.correx.core.sessions.DefaultSessionRepository +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.sessions.projections.replay.EventReplayer +import com.correx.core.transitions.graph.StageConfig +import com.correx.core.transitions.graph.TransitionEdge +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.core.validation.pipeline.ValidationPipeline +import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository +import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore +import com.correx.testing.fixtures.cyclePolicyMissingValidator +import com.correx.testing.fixtures.context.ContextFixtures +import com.correx.testing.fixtures.inference.MockInferenceProvider +import com.correx.testing.fixtures.transitions.TransitionFixtures +import com.correx.testing.kernel.MockSessionEventReplayer +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.yield +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +/** + * Integration test for Task 4.2: the approval gate on stages flagged `requiresApproval`. + * + * Drives analyst→architect with a sequenced stub provider. Asserts that: + * 1. An [OrchestrationPausedEvent] is emitted before the architect runs, with the analyst + * summary in the paired [ApprovalRequestedEvent.preview]. + * 2. Submitting an approval unblocks the run and the workflow completes with the architect's + * execution_plan artifact. + */ +class FreestyleApprovalGateTest { + + private val eventStore = InMemoryEventStore() + private val sessionRepository = DefaultSessionRepository(MockSessionEventReplayer()) + private val orchestrationRepository = OrchestrationRepository( + DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())), + ) + private val inferenceRepository = InferenceRepository( + object : EventReplayer { + override fun rebuild(sessionId: SessionId) = InferenceState() + }, + ) + private val approvalRepository = DefaultApprovalRepository( + DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())), + ) + private val decisionJournalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ) + + private val repositories = OrchestratorRepositories( + eventStore = eventStore, + inferenceRepository = inferenceRepository, + orchestrationRepository = orchestrationRepository, + sessionRepository = sessionRepository, + artifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()), + approvalRepository = approvalRepository, + ) + + private val analysisKind = ConfigArtifactKind( + id = "analysis", + schema = JsonSchema(type = "object", properties = emptyMap()), + llmEmitted = true, + ) + + private val executionPlanKind = ConfigArtifactKind( + id = "execution_plan", + schema = JsonSchema(type = "object", properties = emptyMap()), + llmEmitted = true, + ) + + private val analysisId = ArtifactId("analysis") + private val executionPlanId = ArtifactId("execution_plan") + private val analystStage = StageId("analyst") + private val architectStage = StageId("architect") + + /** Graph mirrors freestyle_planning.toml: architect stage has requiresApproval metadata. */ + private fun freestyleGraph() = WorkflowGraph( + id = "freestyle_planning", + stages = mapOf( + analystStage to StageConfig( + produces = listOf(TypedArtifactSlot(name = analysisId, kind = analysisKind)), + ), + architectStage to StageConfig( + produces = listOf(TypedArtifactSlot(name = executionPlanId, kind = executionPlanKind)), + needs = setOf(analysisId), + metadata = mapOf("requiresApproval" to "true"), + ), + ), + transitions = setOf( + TransitionEdge( + id = TransitionId("analyst-to-architect"), + from = analystStage, + to = architectStage, + condition = { true }, + ), + TransitionEdge( + id = TransitionId("architect-to-done"), + from = architectStage, + to = StageId("done"), + condition = { true }, + ), + ), + start = analystStage, + ) + + private val analysisSummary = """{"summary":"open questions: feasibility of X, timeline for Y"}""" + private val executionPlanJson = """{"plan":"step 1: do A, step 2: do B"}""" + + private fun buildOrchestrator(): DefaultSessionOrchestrator { + // Sequenced provider: first call (analyst) returns the analysis JSON, + // second call (architect) returns the execution plan JSON. + var callCount = 0 + val router = object : com.correx.core.inference.InferenceRouter { + override suspend fun route( + stageId: StageId, + requiredCapabilities: Set, + ) = MockInferenceProvider( + fixedResponse = if (callCount++ == 0) analysisSummary else executionPlanJson, + ) + } + + return DefaultSessionOrchestrator( + repositories = repositories, + engines = OrchestratorEngines( + transitionResolver = TransitionFixtures.simpleResolver(), + contextPackBuilder = ContextFixtures.simpleBuilder(), + inferenceRouter = router, + validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())), + approvalEngine = DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + ), + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = NoopArtifactStore(), + decisionJournalRepository = decisionJournalRepository, + ) + } + + @Test + fun `pause event is emitted with analyst summary before architect runs`(): Unit = runBlocking { + val sessionId = SessionId("freestyle-gate-1") + val orchestrator = buildOrchestrator() + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + + val runJob = launch { orchestrator.run(sessionId, freestyleGraph(), config) } + + // Wait for the pause event before architect + withTimeout(5_000) { + while (eventStore.read(sessionId).none { it.payload is OrchestrationPausedEvent }) { + yield() + } + } + + val events = eventStore.read(sessionId) + val pause = events.firstNotNullOfOrNull { it.payload as? OrchestrationPausedEvent } + assertNotNull(pause, "Expected OrchestrationPausedEvent") + assertTrue(pause!!.stageId == architectStage, "Pause should be for architect stage, got ${pause.stageId}") + assertTrue(pause.reason == "APPROVAL_PENDING", "Expected reason APPROVAL_PENDING, got ${pause.reason}") + + val approvalRequest = events.firstNotNullOfOrNull { it.payload as? ApprovalRequestedEvent } + assertNotNull(approvalRequest, "Expected ApprovalRequestedEvent") + assertTrue( + approvalRequest!!.preview?.contains("open questions") == true || + approvalRequest.preview?.contains("summary") == true, + "Approval preview should carry analyst summary, got: ${approvalRequest.preview}", + ) + + runJob.cancel() + runJob.join() + } + + @Test + fun `approval resumes workflow and architect produces execution plan`(): Unit = runBlocking { + val sessionId = SessionId("freestyle-gate-2") + val orchestrator = buildOrchestrator() + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + + val runJob = launch { orchestrator.run(sessionId, freestyleGraph(), config) } + + // Wait for the approval request + val requestId = withTimeout(5_000) { + var id: com.correx.core.events.types.ApprovalRequestId? = null + while (id == null) { + id = eventStore.read(sessionId) + .firstNotNullOfOrNull { it.payload as? ApprovalRequestedEvent } + ?.requestId + if (id == null) yield() + } + id + } + + val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = architectStage, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val decision = ApprovalDecision( + id = null, + requestId = requestId, + outcome = ApprovalOutcome.APPROVED, + state = ApprovalStatus.COMPLETED, + tier = Tier.T2, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = null, + ) + orchestrator.submitApprovalDecision(requestId, decision) + + runJob.join() + + val events = eventStore.read(sessionId) + assertNotNull( + events.find { it.payload is WorkflowCompletedEvent }, + "Workflow should complete after approval", + ) + } +}