epic-12: after epic audit and init commit
This commit is contained in:
@@ -0,0 +1,300 @@
|
||||
import com.correx.core.approvals.domain.ApprovalEngine
|
||||
import com.correx.core.approvals.domain.DefaultApprovalEngine
|
||||
import com.correx.core.approvals.model.ApprovalContext
|
||||
import com.correx.core.approvals.model.ApprovalGrant
|
||||
import com.correx.core.approvals.model.DomainApprovalRequest
|
||||
import com.correx.core.events.events.OrchestrationPausedEvent
|
||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||
import com.correx.core.events.events.WorkflowCompletedEvent
|
||||
import com.correx.core.events.events.WorkflowFailedEvent
|
||||
import com.correx.core.events.events.WorkflowStartedEvent
|
||||
import com.correx.core.events.execution.RetryPolicy
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.StageId
|
||||
import com.correx.core.inference.InferenceRepository
|
||||
import com.correx.core.inference.InferenceRouter
|
||||
import com.correx.core.inference.InferenceState
|
||||
import com.correx.core.inference.ModelCapability
|
||||
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.WorkflowGraph
|
||||
import com.correx.core.validation.approval.ApprovalTrigger
|
||||
import com.correx.core.validation.pipeline.ValidationPipeline
|
||||
import com.correx.infrastructure.persistence.InMemoryEventStore
|
||||
import com.correx.testing.fixtures.InferenceFixtures
|
||||
import com.correx.testing.fixtures.context.ContextFixtures
|
||||
import com.correx.testing.fixtures.cyclePolicyMissingValidator
|
||||
import com.correx.testing.fixtures.inference.MockInferenceProvider
|
||||
import com.correx.testing.fixtures.transitions.TransitionFixtures
|
||||
import com.correx.testing.fixtures.transitions.TransitionFixtures.threeStageGraph
|
||||
import com.correx.testing.kernel.MockSessionEventReplayer
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.datetime.Instant
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertNotNull
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
@Suppress("UnusedPrivateProperty")
|
||||
class SessionOrchestratorIntegrationTest {
|
||||
private val eventStore = InMemoryEventStore()
|
||||
private val graph = TransitionFixtures.simpleGraph()
|
||||
private val transitionResolver = TransitionFixtures.simpleResolver()
|
||||
private val contextPackBuilder = ContextFixtures.simpleBuilder()
|
||||
private val inferenceRouter = InferenceFixtures.fixedRouter()
|
||||
private val validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator()))
|
||||
private val approvalEngine = DefaultApprovalEngine()
|
||||
private val retryCoordinator = DefaultRetryCoordinator(eventStore)
|
||||
private val sessionReplayer = MockSessionEventReplayer()
|
||||
private val sessionRepository = DefaultSessionRepository(sessionReplayer)
|
||||
val orchestrationReplayer = DefaultEventReplayer(
|
||||
eventStore,
|
||||
OrchestrationProjector(DefaultOrchestrationReducer()),
|
||||
)
|
||||
val orchestrationRepository = OrchestrationRepository(orchestrationReplayer)
|
||||
private val inferenceRepository = InferenceRepository(
|
||||
object : EventReplayer<InferenceState> {
|
||||
override fun rebuild(sessionId: com.correx.core.events.types.SessionId) = InferenceState()
|
||||
},
|
||||
)
|
||||
private val riskAssessor = DefaultRiskAssessor()
|
||||
|
||||
private val repositories = OrchestratorRepositories(
|
||||
eventStore = eventStore,
|
||||
inferenceRepository = inferenceRepository,
|
||||
orchestrationRepository = orchestrationRepository,
|
||||
sessionRepository = sessionRepository,
|
||||
)
|
||||
|
||||
private val engines = OrchestratorEngines(
|
||||
transitionResolver = transitionResolver,
|
||||
contextPackBuilder = contextPackBuilder,
|
||||
inferenceRouter = inferenceRouter,
|
||||
validationPipeline = validationPipeline,
|
||||
approvalEngine = approvalEngine,
|
||||
riskAssessor = riskAssessor,
|
||||
)
|
||||
|
||||
private val orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines,
|
||||
retryCoordinator = retryCoordinator,
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `successful workflow completes all stages`(): Unit = runBlocking {
|
||||
val sessionId = SessionId("s1")
|
||||
val config = OrchestrationConfig(
|
||||
retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0),
|
||||
)
|
||||
|
||||
orchestrator.run(sessionId, graph, config)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
assertEquals(2, events.size)
|
||||
|
||||
val workflowStarted = events.find { it.payload is WorkflowStartedEvent }
|
||||
assertNotNull(workflowStarted)
|
||||
assertEquals(
|
||||
StageId("A"),
|
||||
(workflowStarted?.payload as? WorkflowStartedEvent)?.startStageId,
|
||||
)
|
||||
|
||||
val workflowCompleted = events.find { it.payload is WorkflowCompletedEvent }
|
||||
assertNotNull(workflowCompleted)
|
||||
assertEquals(
|
||||
StageId("B"),
|
||||
(workflowCompleted?.payload as WorkflowCompletedEvent).terminalStageId,
|
||||
)
|
||||
assertEquals(1, (workflowCompleted.payload as WorkflowCompletedEvent).totalStages)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `workflow fails with retry exhaustion`(): Unit = runBlocking {
|
||||
val sessionId = SessionId("s2")
|
||||
val config = OrchestrationConfig(
|
||||
retryPolicy = RetryPolicy(maxAttempts = 2, backoffMs = 0),
|
||||
)
|
||||
val graph = threeStageGraph()
|
||||
|
||||
val failingOrchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines.copy(
|
||||
inferenceRouter = object : InferenceRouter {
|
||||
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>) =
|
||||
MockInferenceProvider(forcedFailure = "Simulated failure")
|
||||
},
|
||||
),
|
||||
retryCoordinator = retryCoordinator,
|
||||
)
|
||||
failingOrchestrator.run(sessionId, graph, config)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val workflowFailed = events.find { it.payload is WorkflowFailedEvent }
|
||||
assertNotNull(workflowFailed)
|
||||
val failed = workflowFailed?.payload as WorkflowFailedEvent
|
||||
assertEquals("Simulated failure", failed.reason)
|
||||
assertTrue(failed.retryExhausted)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `workflow pauses for approval when required`(): Unit = runBlocking {
|
||||
val sessionId = SessionId("s3")
|
||||
val config = OrchestrationConfig(
|
||||
retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0),
|
||||
)
|
||||
val graph = threeStageGraph()
|
||||
val approvingPipeline = ValidationPipeline(
|
||||
validators = listOf(cyclePolicyMissingValidator()),
|
||||
approvalTrigger = ApprovalTrigger(),
|
||||
)
|
||||
val orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines.copy(validationPipeline = approvingPipeline),
|
||||
retryCoordinator = retryCoordinator,
|
||||
)
|
||||
|
||||
orchestrator.run(sessionId, graph, config)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val orPause = events.find { it.payload is OrchestrationPausedEvent }
|
||||
assertNotNull(orPause)
|
||||
val paused = orPause?.let { it.payload as? OrchestrationPausedEvent }
|
||||
assertEquals("APPROVAL_PENDING", paused?.reason)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `workflow resumes after approval`(): Unit = runBlocking {
|
||||
val sessionId = SessionId("s4")
|
||||
val config = OrchestrationConfig(
|
||||
retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0),
|
||||
)
|
||||
val graph = threeStageGraph()
|
||||
val yoloApprovalEngine = object : ApprovalEngine {
|
||||
val inner = DefaultApprovalEngine()
|
||||
override fun evaluate(
|
||||
request: DomainApprovalRequest,
|
||||
context: ApprovalContext,
|
||||
grants: List<ApprovalGrant>,
|
||||
now: Instant,
|
||||
) =
|
||||
inner.evaluate(request, ApprovalContext(context.identity, ApprovalMode.YOLO), grants, now)
|
||||
}
|
||||
|
||||
val approvingPipeline = ValidationPipeline(
|
||||
validators = listOf(cyclePolicyMissingValidator()),
|
||||
approvalTrigger = ApprovalTrigger(),
|
||||
)
|
||||
|
||||
val orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines.copy(
|
||||
validationPipeline = approvingPipeline,
|
||||
approvalEngine = yoloApprovalEngine,
|
||||
),
|
||||
retryCoordinator = retryCoordinator,
|
||||
)
|
||||
|
||||
orchestrator.run(sessionId, graph, config)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val resumeCount = events.count { it.payload is OrchestrationResumedEvent }
|
||||
assertEquals(1, resumeCount)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `workflow with empty graph fails immediately`(): Unit = runBlocking {
|
||||
val emptyGraph = WorkflowGraph(
|
||||
stages = mapOf(StageId("start") to StageConfig()),
|
||||
transitions = emptySet(),
|
||||
start = StageId("start"),
|
||||
)
|
||||
val config = OrchestrationConfig(
|
||||
retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0),
|
||||
)
|
||||
|
||||
val orchestrator = DefaultSessionOrchestrator(
|
||||
repositories = repositories,
|
||||
engines = engines,
|
||||
retryCoordinator = retryCoordinator,
|
||||
)
|
||||
|
||||
val sessionId = SessionId("s6")
|
||||
orchestrator.run(sessionId, emptyGraph, config)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val workflowFailed = events.find { it.payload is WorkflowFailedEvent }
|
||||
assertNotNull(workflowFailed)
|
||||
val failed = workflowFailed?.payload as WorkflowFailedEvent
|
||||
assertNotNull(failed.reason)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `run throws IllegalArgumentException when transition targets undeclared stage`(): Unit = runBlocking {
|
||||
val a = StageId("A")
|
||||
val b = StageId("B")
|
||||
val ghost = StageId("ghost")
|
||||
val z = StageId("Z")
|
||||
// ghost is missing from stages map but has an outgoing transition so it is non-terminal
|
||||
val brokenGraph = WorkflowGraph(
|
||||
stages = mapOf(a to StageConfig(), b to StageConfig()),
|
||||
transitions = setOf(
|
||||
com.correx.core.transitions.graph.TransitionEdge(
|
||||
id = com.correx.core.events.types.TransitionId("t1"), from = a, to = b,
|
||||
condition = { true },
|
||||
),
|
||||
com.correx.core.transitions.graph.TransitionEdge(
|
||||
id = com.correx.core.events.types.TransitionId("t2"), from = b, to = ghost,
|
||||
condition = { true },
|
||||
),
|
||||
com.correx.core.transitions.graph.TransitionEdge(
|
||||
id = com.correx.core.events.types.TransitionId("t3"), from = ghost, to = z,
|
||||
condition = { true },
|
||||
),
|
||||
),
|
||||
start = a,
|
||||
)
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0))
|
||||
val sessionId = SessionId("s7")
|
||||
|
||||
val ex = org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException::class.java) {
|
||||
kotlinx.coroutines.runBlocking { orchestrator.run(sessionId, brokenGraph, config) }
|
||||
}
|
||||
assertTrue(ex.message?.contains("ghost") == true, "Expected message to mention 'ghost', got: ${ex.message}")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `WorkflowStartedEvent carries retryPolicy from config`(): Unit = runBlocking {
|
||||
val sessionId = SessionId("s8")
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 5, backoffMs = 0))
|
||||
|
||||
orchestrator.run(sessionId, graph, config)
|
||||
|
||||
val events = eventStore.read(sessionId)
|
||||
val started = events.first { it.payload is WorkflowStartedEvent }.payload as WorkflowStartedEvent
|
||||
assertEquals(5, started.retryPolicy?.maxAttempts)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `OrchestrationState retryPolicy reflects config after workflow starts`(): Unit = runBlocking {
|
||||
val sessionId = SessionId("s9")
|
||||
val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 7, backoffMs = 0))
|
||||
|
||||
orchestrator.run(sessionId, graph, config)
|
||||
|
||||
val state = orchestrationRepository.getState(sessionId)
|
||||
assertEquals(7, state.retryPolicy?.maxAttempts)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
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.correx.core.transitions.resolution.TransitionOrdering
|
||||
import com.correx.core.validation.approval.ApprovalTrigger
|
||||
import com.correx.core.validation.graph.GraphValidator
|
||||
import com.correx.core.validation.model.ValidationContext
|
||||
import com.correx.core.validation.pipeline.ValidationOutcome
|
||||
import com.correx.core.validation.pipeline.ValidationPipeline
|
||||
import com.correx.core.validation.semantic.SemanticValidator
|
||||
import com.correx.core.validation.semantic.rules.CyclePolicyBindingRule
|
||||
import com.correx.core.validation.transition.TransitionValidator
|
||||
import com.correx.testing.fixtures.CycleFixtures
|
||||
import com.correx.testing.fixtures.WorkflowFixtures
|
||||
import org.junit.jupiter.api.Assertions.assertFalse
|
||||
import org.junit.jupiter.api.Assertions.assertInstanceOf
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class ValidationPipelineIntegrationTest {
|
||||
|
||||
private val pipeline = ValidationPipeline(
|
||||
validators = listOf(
|
||||
GraphValidator(),
|
||||
TransitionValidator(TransitionOrdering.comparator),
|
||||
SemanticValidator(
|
||||
rules = listOf(
|
||||
CyclePolicyBindingRule(requirePolicyForCycles = true),
|
||||
),
|
||||
),
|
||||
),
|
||||
approvalTrigger = ApprovalTrigger()
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `full validation pipeline returns NeedsApproval when cycles are unbound`() {
|
||||
|
||||
val graph = WorkflowFixtures.simpleGraph()
|
||||
|
||||
val cycles = listOf(CycleFixtures.simpleCycle())
|
||||
|
||||
val context = ValidationContext(
|
||||
graph = graph,
|
||||
detectedCycles = cycles,
|
||||
cyclePolicies = emptySet(),
|
||||
)
|
||||
|
||||
val outcome = pipeline.validate(context)
|
||||
|
||||
assertInstanceOf(ValidationOutcome.NeedsApproval::class.java, outcome)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `rejected outcome retryable is false — set by validator, not orchestrator`() {
|
||||
// A dangling transition triggers GraphValidator ERROR → Rejected(retryable=false).
|
||||
// Ownership rule: the validator sets retryable; the orchestrator must only read it.
|
||||
val a = StageId("A")
|
||||
val ghost = StageId("GHOST")
|
||||
val graphWithDanglingTransition = WorkflowGraph(
|
||||
stages = mapOf(a to StageConfig()),
|
||||
transitions = setOf(TransitionEdge(TransitionId("t1"), from = a, to = ghost) { true }),
|
||||
start = a,
|
||||
)
|
||||
val pipeline = ValidationPipeline(validators = listOf(GraphValidator()), approvalTrigger = null)
|
||||
val outcome = pipeline.validate(ValidationContext(graph = graphWithDanglingTransition))
|
||||
|
||||
assertInstanceOf(ValidationOutcome.Rejected::class.java, outcome)
|
||||
assertFalse((outcome as ValidationOutcome.Rejected).retryable)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user