feat(kernel,events): freestyle graph re-routing — deterministic override core
Implements the replay-safe half of preemptive redirect (graph re-routing). An operator-confirmed PreemptRedirectEvent overrides the resolver's edge at the next transition boundary; the pure resolver stays untouched. - PreemptRedirectEvent / PreemptRedirectBlockedEvent (+ serialization registration) - TransitionExecutedEvent + TransitionDecision.Move gain optional redirectId — the durable 'consumed once' marker - PreemptRedirect.decide: pure decision over the event log (override / block / none), so replay reaches the identical outcome without re-classifying (invariant #8) - override wired in DefaultSessionOrchestrator.step above resolveTransition; back-edge jumps count against the existing maxRetries cap via executeMove - blocks jumps to unknown stages or targets with unsatisfied needs - DomainEventMapper: redirect events mapped to null (operator surface ships with the LLM-proposal + approval-confirm front-half) - tests: override+consume-once, block-on-needs, block-on-unknown, ignore-consumed Front-half (LLM proposal -> approval-gate confirm -> emit PreemptRedirectEvent) is deferred; needs live LLM verification. Until it ships no redirect event is emitted in production, so the override is inert. Spec: docs/plans/2026-06-10-freestyle-graph-rerouting.md
This commit is contained in:
@@ -23,6 +23,8 @@ import com.correx.core.events.events.ModelLoadedEvent
|
|||||||
import com.correx.core.events.events.ModelUnloadedEvent
|
import com.correx.core.events.events.ModelUnloadedEvent
|
||||||
import com.correx.core.events.events.OrchestrationPausedEvent
|
import com.correx.core.events.events.OrchestrationPausedEvent
|
||||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
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.StageCompletedEvent
|
||||||
import com.correx.core.events.events.StageFailedEvent
|
import com.correx.core.events.events.StageFailedEvent
|
||||||
import com.correx.core.events.events.StoredEvent
|
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.
|
// Internal slot→CAS-hash bookkeeping (F-007 durable bridge); no operator-facing surface.
|
||||||
is ArtifactContentStoredEvent -> null
|
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 -> {
|
else -> {
|
||||||
log.debug(
|
log.debug(
|
||||||
"DomainEventMapper: unmapped payload type={} sessionId={} sequence={}",
|
"DomainEventMapper: unmapped payload type={} sessionId={} sequence={}",
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -29,5 +29,9 @@ data class TransitionExecutedEvent(
|
|||||||
val sessionId: SessionId,
|
val sessionId: SessionId,
|
||||||
val from: StageId,
|
val from: StageId,
|
||||||
val to: 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
|
) : EventPayload
|
||||||
|
|||||||
@@ -33,6 +33,8 @@ import com.correx.core.events.events.RepoMapComputedEvent
|
|||||||
import com.correx.core.events.events.RetryAttemptedEvent
|
import com.correx.core.events.events.RetryAttemptedEvent
|
||||||
import com.correx.core.events.events.RiskAssessedEvent
|
import com.correx.core.events.events.RiskAssessedEvent
|
||||||
import com.correx.core.events.events.StageCompletedEvent
|
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.StageFailedEvent
|
||||||
import com.correx.core.events.events.SteeringNoteAddedEvent
|
import com.correx.core.events.events.SteeringNoteAddedEvent
|
||||||
import com.correx.core.events.events.ToolCallAssessedEvent
|
import com.correx.core.events.events.ToolCallAssessedEvent
|
||||||
@@ -63,6 +65,8 @@ val eventModule = SerializersModule {
|
|||||||
subclass(ToolCallAssessedEvent::class)
|
subclass(ToolCallAssessedEvent::class)
|
||||||
subclass(FileWrittenEvent::class)
|
subclass(FileWrittenEvent::class)
|
||||||
subclass(SteeringNoteAddedEvent::class)
|
subclass(SteeringNoteAddedEvent::class)
|
||||||
|
subclass(PreemptRedirectEvent::class)
|
||||||
|
subclass(PreemptRedirectBlockedEvent::class)
|
||||||
subclass(InitialIntentEvent::class)
|
subclass(InitialIntentEvent::class)
|
||||||
subclass(StageFailedEvent::class)
|
subclass(StageFailedEvent::class)
|
||||||
subclass(StageCompletedEvent::class)
|
subclass(StageCompletedEvent::class)
|
||||||
|
|||||||
+28
-1
@@ -118,9 +118,36 @@ class DefaultSessionOrchestrator(
|
|||||||
artifactContentCache[cacheKey]?.let { content -> id to content }
|
artifactContentCache[cacheKey]?.let { content -> id to content }
|
||||||
}.toMap()
|
}.toMap()
|
||||||
|
|
||||||
val decision = resolveTransition(
|
val resolved = resolveTransition(
|
||||||
enriched.graph, enriched.sessionId, enriched.currentStageId, stageArtifacts, artifactContent,
|
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(
|
log.debug(
|
||||||
"[Orchestrator] transition session={} stage={} decision={}",
|
"[Orchestrator] transition session={} stage={} decision={}",
|
||||||
enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName,
|
enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName,
|
||||||
|
|||||||
@@ -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<StoredEvent>,
|
||||||
|
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 }
|
||||||
|
}
|
||||||
|
}
|
||||||
+4
-1
@@ -1370,7 +1370,10 @@ abstract class SessionOrchestrator(
|
|||||||
fromStageId: StageId,
|
fromStageId: StageId,
|
||||||
decision: TransitionDecision.Move,
|
decision: TransitionDecision.Move,
|
||||||
): StageId {
|
): StageId {
|
||||||
emit(sessionId, TransitionExecutedEvent(sessionId, fromStageId, decision.to, decision.transitionId))
|
emit(
|
||||||
|
sessionId,
|
||||||
|
TransitionExecutedEvent(sessionId, fromStageId, decision.to, decision.transitionId, decision.redirectId),
|
||||||
|
)
|
||||||
return decision.to
|
return decision.to
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+5
-1
@@ -7,7 +7,11 @@ sealed interface TransitionDecision {
|
|||||||
|
|
||||||
data class Move(
|
data class Move(
|
||||||
val transitionId: TransitionId,
|
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
|
) : TransitionDecision
|
||||||
|
|
||||||
data object Stay : TransitionDecision
|
data object Stay : TransitionDecision
|
||||||
|
|||||||
@@ -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<ContextEntry>,
|
||||||
|
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<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 router = object : InferenceRouter {
|
||||||
|
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>) =
|
||||||
|
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<ArtifactId> = 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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user