diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt index cee9eae2..eab9e4df 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt @@ -23,6 +23,8 @@ import com.correx.core.events.events.ModelLoadedEvent import com.correx.core.events.events.ModelUnloadedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.PreemptRedirectBlockedEvent +import com.correx.core.events.events.PreemptRedirectEvent import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StoredEvent @@ -254,6 +256,11 @@ suspend fun domainEventToServerMessage( // Internal slot→CAS-hash bookkeeping (F-007 durable bridge); no operator-facing surface. is ArtifactContentStoredEvent -> null + // Freestyle graph-rerouting bookkeeping. The deterministic record is in place; a dedicated + // operator surface ships with the LLM-proposal + approval-confirm front-half. + is PreemptRedirectEvent -> null + is PreemptRedirectBlockedEvent -> null + else -> { log.debug( "DomainEventMapper: unmapped payload type={} sessionId={} sequence={}", diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/PreemptEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/PreemptEvents.kt new file mode 100644 index 00000000..588d9e86 --- /dev/null +++ b/core/events/src/main/kotlin/com/correx/core/events/events/PreemptEvents.kt @@ -0,0 +1,45 @@ +package com.correx.core.events.events + +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +/** + * A confirmed preemptive redirect (freestyle graph re-routing): the run should jump from + * [fromStageId] to [toStageId] at the next transition boundary, overriding the resolver's edge. + * + * This is the **recorded, deterministic** half of the hybrid design — an LLM proposes a target + * from a free-text steering note and the operator confirms it (via the approval gate), but only + * the confirmed [toStageId] is recorded here. Replay reads this target and never re-classifies + * (invariants #8/#9). [sourceNoteId] links back to the originating steering note when present. + * + * Consumed exactly once: the override applies it and records [redirectId] on the resulting + * [TransitionExecutedEvent], so a replay sees it already applied. + */ +@Serializable +@SerialName("PreemptRedirect") +data class PreemptRedirectEvent( + val redirectId: String, + val sessionId: SessionId, + val fromStageId: StageId, + val toStageId: StageId, + val sourceNoteId: String? = null, + val timestampMs: Long, +) : EventPayload + +/** + * A preemptive redirect that failed the safety guard — the target stage is unknown/terminal or + * its `needs` artifacts aren't satisfied (decision: invalid jumps are blocked). The run continues + * on its normally-resolved edge. Recording this also marks [redirectId] consumed, so the blocked + * redirect is not retried on every subsequent boundary. + */ +@Serializable +@SerialName("PreemptRedirectBlocked") +data class PreemptRedirectBlockedEvent( + val redirectId: String, + val sessionId: SessionId, + val toStageId: StageId, + val reason: String, + val timestampMs: Long, +) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/StageEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/StageEvents.kt index 0e7300f5..7b70f532 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/StageEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/StageEvents.kt @@ -29,5 +29,9 @@ data class TransitionExecutedEvent( val sessionId: SessionId, val from: StageId, val to: StageId, - val transitionId: TransitionId + val transitionId: TransitionId, + // Set when this transition was produced by a preemptive redirect override (freestyle + // graph re-routing). Carries the originating PreemptRedirectEvent.redirectId — this is the + // durable "redirect consumed" marker the override reads to apply each redirect exactly once. + val redirectId: String? = null, ) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt index 6552ebe6..a63326ce 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt @@ -33,6 +33,8 @@ import com.correx.core.events.events.RepoMapComputedEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.RiskAssessedEvent import com.correx.core.events.events.StageCompletedEvent +import com.correx.core.events.events.PreemptRedirectBlockedEvent +import com.correx.core.events.events.PreemptRedirectEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.ToolCallAssessedEvent @@ -63,6 +65,8 @@ val eventModule = SerializersModule { subclass(ToolCallAssessedEvent::class) subclass(FileWrittenEvent::class) subclass(SteeringNoteAddedEvent::class) + subclass(PreemptRedirectEvent::class) + subclass(PreemptRedirectBlockedEvent::class) subclass(InitialIntentEvent::class) subclass(StageFailedEvent::class) subclass(StageCompletedEvent::class) 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 c2e4ef95..38ac0277 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 @@ -118,9 +118,36 @@ class DefaultSessionOrchestrator( artifactContentCache[cacheKey]?.let { content -> id to content } }.toMap() - val decision = resolveTransition( + val resolved = resolveTransition( enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts, artifactContent, ) + // Preemptive redirect override (freestyle graph re-routing): if the operator confirmed a + // jump (a recorded, unconsumed PreemptRedirectEvent), it overrides the resolver's edge — + // unless the target is unknown or its needs aren't satisfied, in which case it is blocked + // and the normal edge is taken. The decision is pure over the event log (replay-deterministic); + // only the block-event emission is a side effect here. + val decision = when ( + val outcome = PreemptRedirect.decide( + events = repositories.eventStore.read(enriched.sessionId), + graph = enriched.graph, + sessionId = enriched.sessionId, + artifactAvailable = { id -> + !artifactContentCache["${enriched.sessionId.value}:${id.value}"].isNullOrBlank() + }, + nowMs = Clock.System.now().toEpochMilliseconds(), + ) + ) { + is PreemptRedirect.Outcome.Override -> outcome.move + is PreemptRedirect.Outcome.Block -> { + emit(enriched.sessionId, outcome.event) + log.info( + "[Orchestrator] redirect blocked session={} to={} reason={}", + enriched.sessionId.value, outcome.event.toStageId.value, outcome.event.reason, + ) + resolved + } + PreemptRedirect.Outcome.None -> resolved + } log.debug( "[Orchestrator] transition session={} stage={} decision={}", enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/PreemptRedirect.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/PreemptRedirect.kt new file mode 100644 index 00000000..37a041ab --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/PreemptRedirect.kt @@ -0,0 +1,88 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.events.events.PreemptRedirectBlockedEvent +import com.correx.core.events.events.PreemptRedirectEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.TransitionExecutedEvent +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.transitions.graph.WorkflowGraph +import com.correx.core.transitions.resolution.TransitionDecision + +/** + * Pure decision logic for freestyle graph re-routing (the override half). Given the durable event + * log, it decides whether an operator-confirmed [PreemptRedirectEvent] should override the + * resolver's edge. Reads only events + the graph + an artifact-availability predicate, so the + * orchestrator reaches the identical decision on replay (invariant #8). Side effects (emitting the + * block event, executing the move) stay in the orchestrator. + */ +internal object PreemptRedirect { + + sealed interface Outcome { + /** Override the resolver: jump to the confirmed target. */ + data class Override(val move: TransitionDecision.Move) : Outcome + /** Target is unusable (unknown/unsatisfied needs): record this and keep the normal edge. */ + data class Block(val event: PreemptRedirectBlockedEvent) : Outcome + /** No pending redirect — use the resolver's decision unchanged. */ + data object None : Outcome + } + + /** + * @param artifactAvailable true if the given artifact has usable (non-blank) content for this + * session — used to gate the target stage's `needs`. + */ + fun decide( + events: List, + graph: WorkflowGraph, + sessionId: SessionId, + artifactAvailable: (ArtifactId) -> Boolean, + nowMs: Long, + ): Outcome { + val consumed = events.mapNotNull { + when (val p = it.payload) { + is TransitionExecutedEvent -> p.redirectId + is PreemptRedirectBlockedEvent -> p.redirectId + else -> null + } + }.toSet() + val pending = events + .mapNotNull { it.payload as? PreemptRedirectEvent } + .lastOrNull { it.redirectId !in consumed } + ?: return Outcome.None + + val reason = blockReason(graph, pending.toStageId, artifactAvailable) + return if (reason != null) { + Outcome.Block( + PreemptRedirectBlockedEvent( + redirectId = pending.redirectId, + sessionId = sessionId, + toStageId = pending.toStageId, + reason = reason, + timestampMs = nowMs, + ), + ) + } else { + Outcome.Override( + TransitionDecision.Move( + transitionId = TransitionId("redirect:${pending.redirectId}"), + to = pending.toStageId, + redirectId = pending.redirectId, + ), + ) + } + } + + /** Human-readable reason the target is unusable, or null if the jump is valid. */ + private fun blockReason( + graph: WorkflowGraph, + target: StageId, + artifactAvailable: (ArtifactId) -> Boolean, + ): String? { + val stage = graph.stages[target] ?: return "unknown stage '${target.value}'" + val unmet = stage.needs.filterNot(artifactAvailable) + return if (unmet.isEmpty()) null + else "target '${target.value}' has unsatisfied needs: " + unmet.joinToString(", ") { it.value } + } +} 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 fa64af7f..f1dc3eee 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 @@ -1370,7 +1370,10 @@ abstract class SessionOrchestrator( fromStageId: StageId, decision: TransitionDecision.Move, ): StageId { - emit(sessionId, TransitionExecutedEvent(sessionId, fromStageId, decision.to, decision.transitionId)) + emit( + sessionId, + TransitionExecutedEvent(sessionId, fromStageId, decision.to, decision.transitionId, decision.redirectId), + ) return decision.to } diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/resolution/TransitionDecision.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/resolution/TransitionDecision.kt index 668039bb..f4428584 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/resolution/TransitionDecision.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/resolution/TransitionDecision.kt @@ -7,7 +7,11 @@ sealed interface TransitionDecision { data class Move( val transitionId: TransitionId, - val to: StageId + val to: StageId, + // Set only when this Move was produced by a kernel-level preemptive redirect override + // (freestyle graph re-routing). The pure resolver never sets it; it threads to + // TransitionExecutedEvent.redirectId as the "redirect consumed" marker. + val redirectId: String? = null, ) : TransitionDecision data object Stay : TransitionDecision diff --git a/testing/integration/src/test/kotlin/GraphRerouteTest.kt b/testing/integration/src/test/kotlin/GraphRerouteTest.kt new file mode 100644 index 00000000..c2def0dd --- /dev/null +++ b/testing/integration/src/test/kotlin/GraphRerouteTest.kt @@ -0,0 +1,273 @@ +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.context.builder.ContextPackBuilder +import com.correx.core.context.model.ContextEntry +import com.correx.core.context.model.ContextPack +import com.correx.core.context.model.TokenBudget +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.PreemptRedirectBlockedEvent +import com.correx.core.events.events.PreemptRedirectEvent +import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.execution.RetryPolicy +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ContextPackId +import com.correx.core.events.types.EventId +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.InferenceRouter +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.execution.WorkflowResult +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.cyclePolicyMissingValidator +import com.correx.testing.fixtures.inference.MockInferenceProvider +import com.correx.testing.fixtures.transitions.TransitionFixtures +import com.correx.testing.kernel.MockSessionEventReplayer +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test +import java.util.UUID + +/** + * Freestyle graph re-routing — the deterministic override half. A recorded, unconsumed + * [PreemptRedirectEvent] overrides the resolver's edge (jump), unless the target is unknown or its + * `needs` aren't satisfied (block). Exercises the override directly by emitting the confirmed + * redirect event (the front-half's LLM-proposal + approval-confirm is out of scope here). + */ +class GraphRerouteTest { + + private val passthroughBuilder = object : ContextPackBuilder { + override fun build( + id: ContextPackId, + sessionId: SessionId, + stageId: StageId, + entries: List, + budget: TokenBudget, + ): ContextPack = ContextPack( + id = id, + sessionId = sessionId, + stageId = stageId, + layers = emptyMap(), + budgetUsed = 0, + budgetLimit = budget.limit, + ) + } + + 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 router = object : InferenceRouter { + override suspend fun route(stageId: StageId, requiredCapabilities: Set) = + MockInferenceProvider(fixedResponse = "done") + } + + private fun orchestrator() = DefaultSessionOrchestrator( + repositories = repositories, + engines = OrchestratorEngines( + transitionResolver = TransitionFixtures.simpleResolver(), + contextPackBuilder = passthroughBuilder, + inferenceRouter = router, + validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())), + approvalEngine = DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + ), + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = NoopArtifactStore(), + decisionJournalRepository = decisionJournalRepository, + ) + + private val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + + // A → B → C → D, D terminal. All edges always-true, so the unredirected path is A→B→C→D. + private fun linearGraph(cNeeds: Set = emptySet()): WorkflowGraph { + val a = StageId("A"); val b = StageId("B"); val c = StageId("C"); val d = StageId("D") + return WorkflowGraph( + id = "reroute-test", + stages = mapOf( + a to StageConfig(), b to StageConfig(), + c to StageConfig(needs = cNeeds), d to StageConfig(), + ), + transitions = setOf( + TransitionEdge(id = TransitionId("ab"), from = a, to = b, condition = { true }), + TransitionEdge(id = TransitionId("bc"), from = b, to = c, condition = { true }), + TransitionEdge(id = TransitionId("cd"), from = c, to = d, condition = { true }), + ), + start = a, + ) + } + + private suspend fun append(sessionId: SessionId, payload: EventPayload) { + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = payload, + ), + ) + } + + private fun transitions(sessionId: SessionId) = + eventStore.read(sessionId).mapNotNull { it.payload as? TransitionExecutedEvent } + + @Test + fun `confirmed redirect overrides the resolver edge and is consumed once`(): Unit = runBlocking { + val sessionId = SessionId("reroute-jump") + append( + sessionId, + PreemptRedirectEvent( + redirectId = "r1", + sessionId = sessionId, + fromStageId = StageId("A"), + toStageId = StageId("C"), + timestampMs = 1L, + ), + ) + + val result = orchestrator().run(sessionId, linearGraph(), config) + + val ts = transitions(sessionId) + // A jumped straight to C — B was skipped entirely. + assertTrue(ts.none { it.to == StageId("B") }, "B should be skipped, got: ${ts.map { it.to.value }}") + val redirectHop = ts.single { it.redirectId == "r1" } + assertEquals(StageId("C"), redirectHop.to) + assertEquals(StageId("A"), redirectHop.from) + // Consumed exactly once: C advances normally to the terminal D and the run completes, + // rather than re-applying the redirect to C forever. + assertTrue(result is WorkflowResult.Completed, "expected completion, got $result") + assertEquals(StageId("D"), (result as WorkflowResult.Completed).terminalStageId) + } + + @Test + fun `redirect to a target with unsatisfied needs is blocked, normal edge taken`(): Unit = runBlocking { + val sessionId = SessionId("reroute-block-needs") + append( + sessionId, + PreemptRedirectEvent( + redirectId = "r2", + sessionId = sessionId, + fromStageId = StageId("A"), + toStageId = StageId("C"), + timestampMs = 1L, + ), + ) + + // C needs an artifact that is never produced/cached. + orchestrator().run(sessionId, linearGraph(cNeeds = setOf(ArtifactId("design"))), config) + + val blocked = eventStore.read(sessionId).mapNotNull { it.payload as? PreemptRedirectBlockedEvent } + assertEquals(1, blocked.size) + assertTrue(blocked.single().reason.contains("unsatisfied needs"), blocked.single().reason) + // Normal path was taken: the first hop out of A went to B, not C. + val firstFromA = transitions(sessionId).first { it.from == StageId("A") } + assertEquals(StageId("B"), firstFromA.to) + assertNull(firstFromA.redirectId) + } + + @Test + fun `redirect to an unknown stage is blocked`(): Unit = runBlocking { + val sessionId = SessionId("reroute-block-unknown") + append( + sessionId, + PreemptRedirectEvent( + redirectId = "r3", + sessionId = sessionId, + fromStageId = StageId("A"), + toStageId = StageId("ZZZ"), + timestampMs = 1L, + ), + ) + + orchestrator().run(sessionId, linearGraph(), config) + + val blocked = eventStore.read(sessionId).mapNotNull { it.payload as? PreemptRedirectBlockedEvent } + assertEquals(1, blocked.size) + assertTrue(blocked.single().reason.contains("unknown stage"), blocked.single().reason) + } + + @Test + fun `an already-consumed redirect is ignored`(): Unit = runBlocking { + val sessionId = SessionId("reroute-consumed") + // Redirect whose id already appears on a prior TransitionExecuted → treated as applied. + append( + sessionId, + PreemptRedirectEvent( + redirectId = "r4", + sessionId = sessionId, + fromStageId = StageId("A"), + toStageId = StageId("C"), + timestampMs = 1L, + ), + ) + append( + sessionId, + TransitionExecutedEvent(sessionId, StageId("A"), StageId("C"), TransitionId("redirect:r4"), redirectId = "r4"), + ) + + orchestrator().run(sessionId, linearGraph(), config) + + // No block recorded, and the live run resolves A→B normally (the stale redirect is inert). + assertTrue(eventStore.read(sessionId).none { it.payload is PreemptRedirectBlockedEvent }) + val firstFromA = transitions(sessionId).last { it.from == StageId("A") } + assertEquals(StageId("B"), firstFromA.to) + } +}