feat(freestyle): two-phase planning->execution driver (Slice 4)

- freestyle_planning.toml: analyst -> (approval) -> architect, producing
  execution_plan; analyst_freestyle.md surfaces open questions.
- TomlWorkflowLoader: requires_approval stage flag -> metadata.
- SessionOrchestrator: inline-prompt branch (metadata[promptInline]) +
  requestStageApproval helper reusing the existing pause/approve path.
- DefaultSessionOrchestrator: enterStage gates on requiresApproval
  (idempotent via resolved-approval lookup), preview derived from the
  gated stage's needs artifact; validatedArtifactContent accessor.
- FreestyleDriver: compiles validated plan, emits ExecutionPlanLockedEvent,
  runs phase 2 in-session via a runPhase2 seam; wired through ServerModule
  + Main (execution_plan kind registered).
This commit is contained in:
2026-06-08 02:50:36 +04:00
parent 6c8c5e2ad9
commit 37ac767d1f
11 changed files with 744 additions and 17 deletions
@@ -54,8 +54,12 @@ import com.correx.core.toolintent.WorkspacePolicy
import com.correx.core.toolintent.rules.ExecInterpreterRule import com.correx.core.toolintent.rules.ExecInterpreterRule
import com.correx.core.toolintent.rules.NetworkHostRule import com.correx.core.toolintent.rules.NetworkHostRule
import com.correx.core.toolintent.rules.PathContainmentRule 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.InfrastructureModule
import com.correx.infrastructure.inference.DefaultProviderRegistry import com.correx.infrastructure.inference.DefaultProviderRegistry
import com.correx.infrastructure.workflow.ExecutionPlanCompiler
import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy
import com.correx.infrastructure.inference.commons.ManagedInferenceRouter import com.correx.infrastructure.inference.commons.ManagedInferenceRouter
import com.correx.infrastructure.inference.commons.AmdResourceProbe import com.correx.infrastructure.inference.commons.AmdResourceProbe
@@ -306,13 +310,24 @@ fun main() {
} else { } else {
null 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( val module = ServerModule(
orchestrator = orchestrator, orchestrator = orchestrator,
eventStore = eventStore, eventStore = eventStore,
artifactStore = artifactStore, artifactStore = artifactStore,
sessionRepository = repositories.sessionRepository, sessionRepository = repositories.sessionRepository,
workflowRegistry = FileSystemWorkflowRegistry( workflowRegistry = FileSystemWorkflowRegistry(
InfrastructureModule.createWorkflowLoader(loadConfigArtifactKinds(correxConfig)), InfrastructureModule.createWorkflowLoader(configArtifactKinds),
), ),
providerRegistry = infraRegistry.asServerRegistry(), providerRegistry = infraRegistry.asServerRegistry(),
defaultOrchestrationConfig = defaultOrchestrationConfig, defaultOrchestrationConfig = defaultOrchestrationConfig,
@@ -326,6 +341,7 @@ fun main() {
workspaceResolver = workspaceResolver, workspaceResolver = workspaceResolver,
narrationMaxPerRun = routerConfig.narration.maxPerRun, narrationMaxPerRun = routerConfig.narration.maxPerRun,
projectMemory = projectMemory, projectMemory = projectMemory,
freestyleDriver = freestyleDriver,
) )
module.start() module.start()
log.info("==============================") log.info("==============================")
@@ -1,6 +1,7 @@
package com.correx.apps.server package com.correx.apps.server
import com.correx.apps.server.approval.ApprovalCoordinator 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.narration.NarrationSubscriber
import com.correx.apps.server.registry.ProviderRegistry import com.correx.apps.server.registry.ProviderRegistry
import com.correx.apps.server.registry.WorkflowRegistry import com.correx.apps.server.registry.WorkflowRegistry
@@ -69,6 +70,9 @@ class ServerModule(
val narrationMaxPerRun: Int = 100, val narrationMaxPerRun: Int = 100,
// Cross-session, repo-scoped memory. Null disables the hooks entirely (tests / static path). // Cross-session, repo-scoped memory. Null disables the hooks entirely (tests / static path).
val projectMemory: com.correx.apps.server.memory.ProjectMemoryService? = null, 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( val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator(
orchestrator = orchestrator, orchestrator = orchestrator,
@@ -139,6 +143,10 @@ class ServerModule(
} }
runCatching { runCatching {
orchestrator.run(sessionId, graph, sessionConfig) 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. // Distil this run's decisions into durable project memory on completion.
projectMemory?.let { pm -> pm.persist(sessionId, pm.repoRoot()) } projectMemory?.let { pm -> pm.persist(sessionId, pm.repoRoot()) }
}.onFailure { ex -> }.onFailure { ex ->
@@ -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)
}
}
@@ -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<ExecutionPlanLockedEvent>()
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<ExecutionPlanLockedEvent>()
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<ExecutionPlanLockedEvent>()
assertTrue(lockedEvents.isEmpty(), "No lock event when plan content is absent")
assertEquals(0, runPhase2Invocations, "runPhase2 must not be called when plan is absent")
}
}
@@ -8,6 +8,7 @@ import com.correx.core.journal.DefaultDecisionJournalRepository
import com.correx.core.artifacts.ArtifactState import com.correx.core.artifacts.ArtifactState
import com.correx.core.artifactstore.ArtifactStore import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.ApprovalDecisionResolvedEvent 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.ArtifactValidatedEvent
import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.RefinementIterationEvent 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( suspend fun resume(
sessionId: SessionId, sessionId: SessionId,
graph: WorkflowGraph, graph: WorkflowGraph,
@@ -254,6 +259,33 @@ class DefaultSessionOrchestrator(
stageId: StageId, stageId: StageId,
): StepResult { ): StepResult {
log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}") 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) { while (true) {
if (isCancelled(ctx.sessionId)) { if (isCancelled(ctx.sessionId)) {
return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId)) return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId))
@@ -250,21 +250,9 @@ abstract class SessionOrchestrator(
), ),
) )
} ?: emptyList() } ?: emptyList()
val promptEntries = stageConfig.metadata["prompt"] val promptEntries = stageConfig.metadata["promptInline"]
?.let { path -> ?.takeIf { it.isNotBlank() }
val resolvedText = runCatching { promptResolver.resolve(path) } ?.let { text ->
.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( listOf(
ContextEntry( ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()), id = ContextEntryId(UUID.randomUUID().toString()),
@@ -276,7 +264,35 @@ abstract class SessionOrchestrator(
role = EntryRole.USER, 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 } val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted }
require(llmEmittedSlots.size <= 1) { require(llmEmittedSlots.size <= 1) {
@@ -1205,6 +1221,65 @@ abstract class SessionOrchestrator(
return (content.length / 4).coerceAtLeast(1) 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<ApprovalDecision>()
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 functions ---
private suspend fun handleApproval( private suspend fun handleApproval(
@@ -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"
@@ -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.
@@ -39,6 +39,7 @@ private data class StageSection(
@param:JsonProperty("allowed_tools") val allowedTools: List<String> = emptyList(), @param:JsonProperty("allowed_tools") val allowedTools: List<String> = emptyList(),
@param:JsonProperty("token_budget") val tokenBudget: Int = 4096, @param:JsonProperty("token_budget") val tokenBudget: Int = 4096,
@param:JsonProperty("max_retries") val maxRetries: Int = 3, @param:JsonProperty("max_retries") val maxRetries: Int = 3,
@param:JsonProperty("requires_approval") val requiresApproval: Boolean = false,
) )
// condition fields flattened into the transition row // condition fields flattened into the transition row
@@ -94,6 +95,7 @@ class TomlWorkflowLoader(
metadata = buildMap { metadata = buildMap {
s.prompt?.let { put("prompt", resolvePromptPath(it, workflowDir)) } s.prompt?.let { put("prompt", resolvePromptPath(it, workflowDir)) }
s.systemPrompt?.let { put("systemPrompt", resolvePromptPath(it, workflowDir)) } s.systemPrompt?.let { put("systemPrompt", resolvePromptPath(it, workflowDir)) }
if (s.requiresApproval) put("requiresApproval", "true")
}, },
) )
} }
@@ -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)
}
}
@@ -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<InferenceState> {
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<ModelCapability>,
) = 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",
)
}
}