From fac6a29178cc394dbb389c34c0dcc6433dd2a746 Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 4 Jun 2026 00:54:37 +0400 Subject: [PATCH] feat(kernel): runtime refinement guard on back-edges with iteration cap --- .../DefaultSessionOrchestrator.kt | 30 ++++ .../src/test/kotlin/RefinementLoopTest.kt | 155 ++++++++++++++++++ 2 files changed, 185 insertions(+) create mode 100644 testing/integration/src/test/kotlin/RefinementLoopTest.kt 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 36d59b86..b3dd4c61 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 @@ -10,6 +10,8 @@ import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.ApprovalDecisionResolvedEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.RefinementIterationEvent +import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.types.ApprovalDecisionId import com.correx.core.events.types.ApprovalRequestId @@ -35,6 +37,9 @@ import java.util.concurrent.atomic.* ) private val log = LoggerFactory.getLogger(DefaultSessionOrchestrator::class.java) +// Fallback refinement-loop cap when a stage does not declare maxRetries. +private const val DEFAULT_MAX_REFINEMENT = 3 + class DefaultSessionOrchestrator( private val repositories: OrchestratorRepositories, engines: OrchestratorEngines, @@ -206,6 +211,27 @@ class DefaultSessionOrchestrator( ) } + // Runtime refinement guard: a back-edge (re-entering a stage already visited this + // run, e.g. reviewer→implementer) increments a per-cycle counter recorded as an + // event, so the guard is replay-deterministic. Exceeding the cap escalates to a + // terminal failure instead of looping forever. + if (isBackEdge(ctx, nextStageId)) { + val cycleKey = "${ctx.currentStageId.value}->${nextStageId.value}" + val maxIterations = ctx.graph.stages[nextStageId]?.maxRetries ?: DEFAULT_MAX_REFINEMENT + val iteration = (orchestrationRepository.getState(ctx.sessionId).refinementIterations[cycleKey] ?: 0) + 1 + emit(ctx.sessionId, RefinementIterationEvent(ctx.sessionId, cycleKey, iteration, maxIterations)) + if (iteration > maxIterations) { + return StepResult.Terminal( + failWorkflow( + ctx.sessionId, + nextStageId, + "refinement loop '$cycleKey' exceeded $maxIterations iterations — escalating", + retryExhausted = true, + ), + ) + } + } + // Emit the transition before executing the next stage. Otherwise the next stage's // events (InferenceStarted, artifacts, …) are recorded ahead of the TransitionExecuted // that marks entering it, so the log shows the stage running before it was entered. @@ -265,6 +291,10 @@ class DefaultSessionOrchestrator( } } + private fun isBackEdge(ctx: EnrichedExecutionContext, target: StageId): Boolean = + repositories.eventStore.read(ctx.sessionId) + .any { (it.payload as? TransitionExecutedEvent)?.to == target } + private fun ExecutionContext.enrich() = EnrichedExecutionContext( graph, sessionId, stageCount, currentStageId, config, session = repositories.sessionRepository.getSession(sessionId), diff --git a/testing/integration/src/test/kotlin/RefinementLoopTest.kt b/testing/integration/src/test/kotlin/RefinementLoopTest.kt new file mode 100644 index 00000000..86b94b99 --- /dev/null +++ b/testing/integration/src/test/kotlin/RefinementLoopTest.kt @@ -0,0 +1,155 @@ +import com.correx.core.approvals.ApprovalProjector +import com.correx.core.approvals.DefaultApprovalReducer +import com.correx.core.approvals.DefaultApprovalRepository +import com.correx.core.approvals.domain.DefaultApprovalEngine +import com.correx.core.artifacts.DefaultArtifactReducer +import com.correx.core.events.events.RefinementIterationEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowFailedEvent +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.events.types.TransitionId +import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.InferenceState +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.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.InferenceFixtures +import com.correx.testing.fixtures.context.ContextFixtures +import com.correx.testing.fixtures.cyclePolicyMissingValidator +import com.correx.testing.fixtures.transitions.TransitionFixtures +import com.correx.testing.kernel.MockSessionEventReplayer +import kotlinx.coroutines.runBlocking +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 + +class RefinementLoopTest { + 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 engines = OrchestratorEngines( + transitionResolver = TransitionFixtures.simpleResolver(), + contextPackBuilder = ContextFixtures.simpleBuilder(), + inferenceRouter = InferenceFixtures.fixedRouter(), + validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())), + approvalEngine = DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + ) + + private val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = NoopArtifactStore(), + decisionJournalRepository = decisionJournalRepository, + ) + + // A→B and B→A, both unconditional: an unbounded loop absent the runtime guard. + // B declares maxRetries=2 → the "A->B" cycle is capped at 2 iterations. + private fun loopGraph() = WorkflowGraph( + id = "refine-loop", + stages = mapOf( + StageId("A") to StageConfig(), + StageId("B") to StageConfig(maxRetries = 2), + ), + transitions = setOf( + TransitionEdge(TransitionId("t-ab"), StageId("A"), StageId("B"), condition = { true }), + TransitionEdge(TransitionId("t-ba"), StageId("B"), StageId("A"), condition = { true }), + ), + start = StageId("A"), + ) + + @Test + fun `guard terminates an unbounded loop at maxIterations`(): Unit = runBlocking { + val sessionId = SessionId("loop-1") + orchestrator.run( + sessionId, + loopGraph(), + OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)), + ) + + val events = eventStore.read(sessionId) + val iterations = events.mapNotNull { it.payload as? RefinementIterationEvent } + assertTrue(iterations.isNotEmpty(), "expected RefinementIterationEvent on back-edges") + + val failed = events.mapNotNull { it.payload as? WorkflowFailedEvent } + assertTrue(failed.isNotEmpty(), "guard must escalate to a terminal failure") + assertTrue( + failed.first().reason.contains("refinement loop"), + "unexpected failure reason: ${failed.first().reason}", + ) + // The "A->B" cycle is capped at 2 → it fires the terminal failure on iteration 3. + val ab = iterations.filter { it.cycleKey == "A->B" } + assertEquals(3, ab.maxOf { it.iteration }) + } + + @Test + fun `forward-only workflow emits no refinement iteration`(): Unit = runBlocking { + val sessionId = SessionId("forward-1") + val graph = WorkflowGraph( + id = "forward", + stages = mapOf(StageId("A") to StageConfig()), + transitions = setOf( + TransitionEdge(TransitionId("t1"), StageId("A"), StageId("done"), condition = { true }), + ), + start = StageId("A"), + ) + orchestrator.run( + sessionId, + graph, + OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)), + ) + + val events = eventStore.read(sessionId) + assertNotNull(events.find { it.payload is WorkflowCompletedEvent }) + assertTrue( + events.none { it.payload is RefinementIterationEvent }, + "forward edges must not trigger the refinement guard", + ) + } +}