kami
4d1b4ffbb6
- DecisionKind gains PREEMPT; DecisionJournalState tracks planLocked. - DefaultDecisionJournalReducer: ExecutionPlanLockedEvent flips planLocked (no record); post-lock steering notes recorded as PREEMPT with a 'PRIORITY DIRECTIVE' summary, pre-lock stay STEERING. - SessionOrchestrator.buildSteeringNoteEntries: post-lock notes surfaced as pinned L0/SYSTEM 'PRIORITY DIRECTIVE' entries ahead of the plan; pre-lock keep L2. Graph re-routing intentionally deferred (priority surfacing only). - Tests: PreemptSteeringJournalTest, SteeringPreemptionTest.
253 lines
11 KiB
Kotlin
253 lines
11 KiB
Kotlin
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.ContextLayer
|
|
import com.correx.core.context.model.ContextPack
|
|
import com.correx.core.context.model.TokenBudget
|
|
import com.correx.core.events.events.ExecutionPlanLockedEvent
|
|
import com.correx.core.events.events.EventMetadata
|
|
import com.correx.core.events.events.NewEvent
|
|
import com.correx.core.events.events.SteeringNoteAddedEvent
|
|
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.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.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.assertTrue
|
|
import org.junit.jupiter.api.Test
|
|
import java.util.UUID
|
|
|
|
/**
|
|
* Verifies that a [SteeringNoteAddedEvent] emitted after an [ExecutionPlanLockedEvent] is
|
|
* injected into the stage context as a pinned L0/SYSTEM "PRIORITY DIRECTIVE" entry, while
|
|
* a pre-lock steering note retains the original L2/SYSTEM form.
|
|
*/
|
|
class SteeringPreemptionTest {
|
|
|
|
private val capturedEntries = mutableListOf<ContextEntry>()
|
|
|
|
private val capturingBuilder = object : ContextPackBuilder {
|
|
override fun build(
|
|
id: ContextPackId,
|
|
sessionId: SessionId,
|
|
stageId: StageId,
|
|
entries: List<ContextEntry>,
|
|
budget: TokenBudget,
|
|
): ContextPack {
|
|
capturedEntries.addAll(entries)
|
|
return ContextPack(
|
|
id = id,
|
|
sessionId = sessionId,
|
|
stageId = stageId,
|
|
layers = emptyMap(),
|
|
budgetUsed = 0,
|
|
budgetLimit = budget.limit,
|
|
)
|
|
}
|
|
}
|
|
|
|
private val eventStore = InMemoryEventStore()
|
|
private val sessionReplayer = MockSessionEventReplayer()
|
|
private val sessionRepository = DefaultSessionRepository(sessionReplayer)
|
|
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 artifactStore = NoopArtifactStore()
|
|
private val retryCoordinator = DefaultRetryCoordinator(eventStore)
|
|
|
|
private val repositories = OrchestratorRepositories(
|
|
eventStore = eventStore,
|
|
inferenceRepository = inferenceRepository,
|
|
orchestrationRepository = orchestrationRepository,
|
|
sessionRepository = sessionRepository,
|
|
artifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()),
|
|
approvalRepository = approvalRepository,
|
|
)
|
|
|
|
private fun buildOrchestrator(inferenceRouter: com.correx.core.inference.InferenceRouter) =
|
|
DefaultSessionOrchestrator(
|
|
repositories = repositories,
|
|
engines = OrchestratorEngines(
|
|
transitionResolver = TransitionFixtures.simpleResolver(),
|
|
contextPackBuilder = capturingBuilder,
|
|
inferenceRouter = inferenceRouter,
|
|
validationPipeline = ValidationPipeline(validators = listOf(cyclePolicyMissingValidator())),
|
|
approvalEngine = DefaultApprovalEngine(),
|
|
riskAssessor = DefaultRiskAssessor(),
|
|
),
|
|
retryCoordinator = retryCoordinator,
|
|
artifactStore = artifactStore,
|
|
decisionJournalRepository = decisionJournalRepository,
|
|
)
|
|
|
|
private val defaultConfig = OrchestrationConfig(
|
|
retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0),
|
|
)
|
|
|
|
private suspend fun appendEvent(sessionId: SessionId, payload: com.correx.core.events.events.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,
|
|
),
|
|
)
|
|
}
|
|
|
|
@Test
|
|
fun `post-lock steering note is injected as L0 PRIORITY DIRECTIVE`(): Unit = runBlocking {
|
|
val sessionId = SessionId("steering-preemption-1")
|
|
val stageA = StageId("A")
|
|
|
|
val graph = WorkflowGraph(
|
|
id = "steering-preemption-test",
|
|
stages = mapOf(stageA to StageConfig()),
|
|
transitions = emptySet(),
|
|
start = stageA,
|
|
)
|
|
|
|
appendEvent(
|
|
sessionId,
|
|
ExecutionPlanLockedEvent(
|
|
sessionId = sessionId,
|
|
planArtifactId = ArtifactId("plan"),
|
|
workflowId = "steering-preemption-test",
|
|
stageIds = listOf(stageA.value),
|
|
startStageId = stageA.value,
|
|
),
|
|
)
|
|
appendEvent(
|
|
sessionId,
|
|
SteeringNoteAddedEvent(
|
|
sessionId = sessionId,
|
|
content = "use postgres not sqlite",
|
|
),
|
|
)
|
|
|
|
val router = object : com.correx.core.inference.InferenceRouter {
|
|
override suspend fun route(
|
|
stageId: StageId,
|
|
requiredCapabilities: Set<ModelCapability>,
|
|
) = MockInferenceProvider(fixedResponse = "done")
|
|
}
|
|
|
|
buildOrchestrator(router).run(sessionId, graph, defaultConfig)
|
|
|
|
val steeringEntries = capturedEntries.filter { it.sourceType == "steeringNote" }
|
|
assertTrue(steeringEntries.isNotEmpty(), "Expected at least one steeringNote entry")
|
|
val postLockEntry = steeringEntries.find { it.content.contains("PRIORITY DIRECTIVE") }
|
|
assertTrue(
|
|
postLockEntry != null,
|
|
"Expected a steeringNote entry with 'PRIORITY DIRECTIVE', got: ${steeringEntries.map { it.content }}",
|
|
)
|
|
assertTrue(
|
|
postLockEntry!!.content.contains("use postgres not sqlite"),
|
|
"Expected steering text in entry, got: ${postLockEntry.content}",
|
|
)
|
|
assertTrue(
|
|
postLockEntry.layer == ContextLayer.L0,
|
|
"Expected post-lock steering note at L0, got: ${postLockEntry.layer}",
|
|
)
|
|
}
|
|
|
|
@Test
|
|
fun `pre-lock steering note keeps L2 SYSTEM form`(): Unit = runBlocking {
|
|
val sessionId = SessionId("steering-preemption-2")
|
|
val stageA = StageId("A")
|
|
|
|
val graph = WorkflowGraph(
|
|
id = "steering-preemption-test-2",
|
|
stages = mapOf(stageA to StageConfig()),
|
|
transitions = emptySet(),
|
|
start = stageA,
|
|
)
|
|
|
|
appendEvent(
|
|
sessionId,
|
|
SteeringNoteAddedEvent(
|
|
sessionId = sessionId,
|
|
content = "prefer immutable data",
|
|
),
|
|
)
|
|
// No ExecutionPlanLockedEvent — note is pre-lock
|
|
|
|
val router = object : com.correx.core.inference.InferenceRouter {
|
|
override suspend fun route(
|
|
stageId: StageId,
|
|
requiredCapabilities: Set<ModelCapability>,
|
|
) = MockInferenceProvider(fixedResponse = "done")
|
|
}
|
|
|
|
buildOrchestrator(router).run(sessionId, graph, defaultConfig)
|
|
|
|
val steeringEntries = capturedEntries.filter { it.sourceType == "steeringNote" }
|
|
assertTrue(steeringEntries.isNotEmpty(), "Expected at least one steeringNote entry")
|
|
val preLockEntry = steeringEntries.find { it.content.contains("prefer immutable data") }
|
|
assertTrue(
|
|
preLockEntry != null,
|
|
"Expected a steeringNote entry with pre-lock content",
|
|
)
|
|
assertTrue(
|
|
!preLockEntry!!.content.contains("PRIORITY DIRECTIVE"),
|
|
"Pre-lock steering note must not contain 'PRIORITY DIRECTIVE', got: ${preLockEntry.content}",
|
|
)
|
|
assertTrue(
|
|
preLockEntry.layer == ContextLayer.L2,
|
|
"Expected pre-lock steering note at L2, got: ${preLockEntry.layer}",
|
|
)
|
|
}
|
|
}
|