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.
This commit is contained in:
+5
-2
@@ -2,6 +2,7 @@ package com.correx.core.journal
|
|||||||
|
|
||||||
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
||||||
import com.correx.core.events.events.ArtifactValidatedEvent
|
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.InitialIntentEvent
|
||||||
import com.correx.core.events.events.RetryAttemptedEvent
|
import com.correx.core.events.events.RetryAttemptedEvent
|
||||||
import com.correx.core.events.events.StageFailedEvent
|
import com.correx.core.events.events.StageFailedEvent
|
||||||
@@ -14,6 +15,7 @@ import com.correx.core.journal.model.DecisionRecord
|
|||||||
|
|
||||||
class DefaultDecisionJournalReducer : DecisionJournalReducer {
|
class DefaultDecisionJournalReducer : DecisionJournalReducer {
|
||||||
override fun reduce(state: DecisionJournalState, event: StoredEvent): DecisionJournalState {
|
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) {
|
val record = when (val p = event.payload) {
|
||||||
is InitialIntentEvent -> DecisionRecord(
|
is InitialIntentEvent -> DecisionRecord(
|
||||||
sequence = event.sessionSequence,
|
sequence = event.sessionSequence,
|
||||||
@@ -22,8 +24,9 @@ class DefaultDecisionJournalReducer : DecisionJournalReducer {
|
|||||||
)
|
)
|
||||||
is SteeringNoteAddedEvent -> DecisionRecord(
|
is SteeringNoteAddedEvent -> DecisionRecord(
|
||||||
sequence = event.sessionSequence,
|
sequence = event.sessionSequence,
|
||||||
kind = DecisionKind.STEERING,
|
kind = if (state.planLocked) DecisionKind.PREEMPT else DecisionKind.STEERING,
|
||||||
summary = "User steering: ${p.content}",
|
summary = if (state.planLocked) "PRIORITY DIRECTIVE (overrides plan): ${p.content}"
|
||||||
|
else "User steering: ${p.content}",
|
||||||
stageId = p.stageId?.value,
|
stageId = p.stageId?.value,
|
||||||
)
|
)
|
||||||
is ApprovalDecisionResolvedEvent -> DecisionRecord(
|
is ApprovalDecisionResolvedEvent -> DecisionRecord(
|
||||||
|
|||||||
@@ -5,4 +5,5 @@ import kotlinx.serialization.Serializable
|
|||||||
@Serializable
|
@Serializable
|
||||||
data class DecisionJournalState(
|
data class DecisionJournalState(
|
||||||
val records: List<DecisionRecord> = emptyList(),
|
val records: List<DecisionRecord> = emptyList(),
|
||||||
|
val planLocked: Boolean = false,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ package com.correx.core.journal.model
|
|||||||
|
|
||||||
import kotlinx.serialization.Serializable
|
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
|
* One distilled decision. [sequence] is the source event's sessionSequence, used for
|
||||||
|
|||||||
+6
-2
@@ -38,6 +38,7 @@ import com.correx.core.events.events.NewEvent
|
|||||||
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.RiskAssessedEvent
|
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.SteeringNoteAddedEvent
|
||||||
import com.correx.core.events.events.ToolExecutionRejectedEvent
|
import com.correx.core.events.events.ToolExecutionRejectedEvent
|
||||||
import com.correx.core.events.events.ToolInvocationRequestedEvent
|
import com.correx.core.events.events.ToolInvocationRequestedEvent
|
||||||
@@ -787,12 +788,15 @@ abstract class SessionOrchestrator(
|
|||||||
|
|
||||||
private suspend fun buildSteeringNoteEntries(sessionId: SessionId): List<ContextEntry> {
|
private suspend fun buildSteeringNoteEntries(sessionId: SessionId): List<ContextEntry> {
|
||||||
val events = eventStore.read(sessionId)
|
val events = eventStore.read(sessionId)
|
||||||
|
var locked = false
|
||||||
return events.mapNotNull { event ->
|
return events.mapNotNull { event ->
|
||||||
when (val p = event.payload) {
|
when (val p = event.payload) {
|
||||||
|
is ExecutionPlanLockedEvent -> { locked = true; null }
|
||||||
is SteeringNoteAddedEvent -> ContextEntry(
|
is SteeringNoteAddedEvent -> ContextEntry(
|
||||||
id = ContextEntryId(UUID.randomUUID().toString()),
|
id = ContextEntryId(UUID.randomUUID().toString()),
|
||||||
layer = ContextLayer.L2,
|
layer = if (locked) ContextLayer.L0 else ContextLayer.L2,
|
||||||
content = p.content,
|
content = if (locked) "PRIORITY DIRECTIVE — overrides the locked plan: ${p.content}"
|
||||||
|
else p.content,
|
||||||
sourceType = "steeringNote",
|
sourceType = "steeringNote",
|
||||||
sourceId = p.stageId?.value ?: sessionId.value,
|
sourceId = p.stageId?.value ?: sessionId.value,
|
||||||
tokenEstimate = estimateTokens(p.content),
|
tokenEstimate = estimateTokens(p.content),
|
||||||
|
|||||||
@@ -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<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}",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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}",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user