From 4d1b4ffbb680951efe4c6c6ec3fb9f35a11b6b73 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 8 Jun 2026 02:58:21 +0400 Subject: [PATCH] feat(freestyle): soft-lock steering preemption (Slice 5) - 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. --- .../journal/DefaultDecisionJournalReducer.kt | 7 +- .../journal/model/DecisionJournalState.kt | 1 + .../core/journal/model/DecisionRecord.kt | 2 +- .../orchestration/SessionOrchestrator.kt | 8 +- .../src/test/kotlin/SteeringPreemptionTest.kt | 252 ++++++++++++++++++ .../test/kotlin/PreemptSteeringJournalTest.kt | 49 ++++ 6 files changed, 314 insertions(+), 5 deletions(-) create mode 100644 testing/integration/src/test/kotlin/SteeringPreemptionTest.kt create mode 100644 testing/projections/src/test/kotlin/PreemptSteeringJournalTest.kt diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt b/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt index c48d07ce..a5d79485 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/DefaultDecisionJournalReducer.kt @@ -2,6 +2,7 @@ package com.correx.core.journal import com.correx.core.events.events.ApprovalDecisionResolvedEvent import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.ExecutionPlanLockedEvent import com.correx.core.events.events.InitialIntentEvent import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.StageFailedEvent @@ -14,6 +15,7 @@ import com.correx.core.journal.model.DecisionRecord class DefaultDecisionJournalReducer : DecisionJournalReducer { override fun reduce(state: DecisionJournalState, event: StoredEvent): DecisionJournalState { + if (event.payload is ExecutionPlanLockedEvent) return state.copy(planLocked = true) val record = when (val p = event.payload) { is InitialIntentEvent -> DecisionRecord( sequence = event.sessionSequence, @@ -22,8 +24,9 @@ class DefaultDecisionJournalReducer : DecisionJournalReducer { ) is SteeringNoteAddedEvent -> DecisionRecord( sequence = event.sessionSequence, - kind = DecisionKind.STEERING, - summary = "User steering: ${p.content}", + kind = if (state.planLocked) DecisionKind.PREEMPT else DecisionKind.STEERING, + summary = if (state.planLocked) "PRIORITY DIRECTIVE (overrides plan): ${p.content}" + else "User steering: ${p.content}", stageId = p.stageId?.value, ) is ApprovalDecisionResolvedEvent -> DecisionRecord( diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionJournalState.kt b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionJournalState.kt index 4dae04b1..21365704 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionJournalState.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionJournalState.kt @@ -5,4 +5,5 @@ import kotlinx.serialization.Serializable @Serializable data class DecisionJournalState( val records: List = emptyList(), + val planLocked: Boolean = false, ) diff --git a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt index 58809ae7..4be96a04 100644 --- a/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt +++ b/core/journal/src/main/kotlin/com/correx/core/journal/model/DecisionRecord.kt @@ -2,7 +2,7 @@ package com.correx.core.journal.model import kotlinx.serialization.Serializable -enum class DecisionKind { INTENT, STEERING, APPROVAL, TRANSITION, RETRY, FAILURE, ARTIFACT } +enum class DecisionKind { INTENT, STEERING, PREEMPT, APPROVAL, TRANSITION, RETRY, FAILURE, ARTIFACT } /** * One distilled decision. [sequence] is the source event's sessionSequence, used for 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 a55ce511..dd1f8816 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 @@ -38,6 +38,7 @@ import com.correx.core.events.events.NewEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RiskAssessedEvent +import com.correx.core.events.events.ExecutionPlanLockedEvent import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.ToolExecutionRejectedEvent import com.correx.core.events.events.ToolInvocationRequestedEvent @@ -787,12 +788,15 @@ abstract class SessionOrchestrator( private suspend fun buildSteeringNoteEntries(sessionId: SessionId): List { val events = eventStore.read(sessionId) + var locked = false return events.mapNotNull { event -> when (val p = event.payload) { + is ExecutionPlanLockedEvent -> { locked = true; null } is SteeringNoteAddedEvent -> ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), - layer = ContextLayer.L2, - content = p.content, + layer = if (locked) ContextLayer.L0 else ContextLayer.L2, + content = if (locked) "PRIORITY DIRECTIVE — overrides the locked plan: ${p.content}" + else p.content, sourceType = "steeringNote", sourceId = p.stageId?.value ?: sessionId.value, tokenEstimate = estimateTokens(p.content), diff --git a/testing/integration/src/test/kotlin/SteeringPreemptionTest.kt b/testing/integration/src/test/kotlin/SteeringPreemptionTest.kt new file mode 100644 index 00000000..4bb3195a --- /dev/null +++ b/testing/integration/src/test/kotlin/SteeringPreemptionTest.kt @@ -0,0 +1,252 @@ +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() + + private val capturingBuilder = object : ContextPackBuilder { + override fun build( + id: ContextPackId, + sessionId: SessionId, + stageId: StageId, + entries: List, + 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 { + 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, + ) = 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, + ) = 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}", + ) + } +} diff --git a/testing/projections/src/test/kotlin/PreemptSteeringJournalTest.kt b/testing/projections/src/test/kotlin/PreemptSteeringJournalTest.kt new file mode 100644 index 00000000..24adab02 --- /dev/null +++ b/testing/projections/src/test/kotlin/PreemptSteeringJournalTest.kt @@ -0,0 +1,49 @@ +import com.correx.core.events.events.ExecutionPlanLockedEvent +import com.correx.core.events.events.SteeringNoteAddedEvent +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.SessionId +import com.correx.core.journal.DecisionJournalProjector +import com.correx.core.journal.DefaultDecisionJournalReducer +import com.correx.core.journal.model.DecisionKind +import com.correx.testing.fixtures.EventFixtures.stored +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class PreemptSteeringJournalTest { + private val projector = DecisionJournalProjector(DefaultDecisionJournalReducer()) + + @Test + fun `steering before plan lock is STEERING, steering after is PREEMPT`() { + var s = projector.initial() + s = projector.apply( + s, + stored(payload = SteeringNoteAddedEvent(SessionId("x"), "use jwt"), sequence = 1L), + ) + s = projector.apply( + s, + stored( + payload = ExecutionPlanLockedEvent( + sessionId = SessionId("x"), + planArtifactId = ArtifactId("plan-1"), + workflowId = "wf1", + stageIds = listOf("stage1"), + startStageId = "stage1", + ), + sequence = 2L, + ), + ) + s = projector.apply( + s, + stored(payload = SteeringNoteAddedEvent(SessionId("x"), "skip tests"), sequence = 3L), + ) + + assertEquals(2, s.records.size) + assertEquals(DecisionKind.STEERING, s.records[0].kind) + assertEquals(DecisionKind.PREEMPT, s.records[1].kind) + assertTrue( + s.records[1].summary.startsWith("PRIORITY DIRECTIVE"), + "Expected summary to start with 'PRIORITY DIRECTIVE' but was: ${s.records[1].summary}", + ) + } +}