feat(kernel): SubagentRunner seam for stage dispatch

Extract a SubagentRunner fun-interface (SubagentRunRequest/SubagentRunResult)
and an InSessionSubagentRunner default that delegates to the existing
executeStage path, with per-run effectives captured in the lambda closure so
the seam stays free of the internal RunEffectives type.

DefaultSessionOrchestrator.enterStage now dispatches through subagentRunner.run
instead of calling executeStage inline; effectives threading is dropped from
run/step/executeMove/enterStage/resume. ReplayOrchestrator supplies its own
runner. Behaviour is identical — RefinementLoopTest and full check stay green.

core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt

@@ -20,6 +20,9 @@ import com.correx.core.events.types.ArtifactLifecyclePhase
 import com.correx.core.events.types.SessionId
 import com.correx.core.events.types.StageId
 import com.correx.core.kernel.execution.WorkflowResult
+import com.correx.core.kernel.orchestration.subagent.InSessionSubagentRunner
+import com.correx.core.kernel.orchestration.subagent.SubagentRunRequest
+import com.correx.core.kernel.orchestration.subagent.SubagentRunner
 import com.correx.core.kernel.retry.RetryCoordinator
 import com.correx.core.sessions.Session
 import com.correx.core.transitions.execution.StageExecutionResult
@@ -52,12 +55,17 @@ class DefaultSessionOrchestrator(
     override val cancellations: ConcurrentHashMap<SessionId, AtomicBoolean> =
         ConcurrentHashMap<SessionId, AtomicBoolean>()
 
+    override val subagentRunner: SubagentRunner = InSessionSubagentRunner(
+        executeStage = { sid, stg, graph, session, cfg ->
+            executeStage(sid, stg, graph, session, cfg, effectivesFor(cfg))
+        },
+    )
+
     override suspend fun run(
         sessionId: SessionId,
         graph: WorkflowGraph,
         config: OrchestrationConfig,
     ): WorkflowResult {
-        val effectives = effectivesFor(config)
         log.debug("[Orchestrator] session={} workflow={} start={}", sessionId.value, graph.id, graph.start.value)
         emitWorkflowStarted(sessionId, graph, config)
 
@@ -65,14 +73,14 @@ class DefaultSessionOrchestrator(
         val enriched = base.enrich()
 
         // Execute the start stage before entering the step loop
-        return when (val result = enterStage(enriched, graph.start, effectives)) {
-            is StepResult.Continue -> step(result.ctx, effectives)
+        return when (val result = enterStage(enriched, graph.start)) {
+            is StepResult.Continue -> step(result.ctx)
             is StepResult.Terminal -> result.result
         }
     }
 
     @Suppress("LongMethod")
-    private tailrec suspend fun step(ctx: EnrichedExecutionContext, effectives: RunEffectives): WorkflowResult {
+    private tailrec suspend fun step(ctx: EnrichedExecutionContext): WorkflowResult {
         log.debug(
             "[Orchestrator] step session={} stage={} stageCount={}",
             ctx.sessionId.value, ctx.currentStageId.value, ctx.stageCount,
@@ -114,8 +122,8 @@ class DefaultSessionOrchestrator(
             enriched.sessionId.value, enriched.currentStageId.value, decision::class.simpleName,
         )
         return when (decision) {
-            is TransitionDecision.Move -> when (val r = executeMove(enriched, decision, effectives)) {
-                is StepResult.Continue -> step(r.ctx, effectives)
+            is TransitionDecision.Move -> when (val r = executeMove(enriched, decision)) {
+                is StepResult.Continue -> step(r.ctx)
                 is StepResult.Terminal -> r.result
             }
 
@@ -152,14 +160,13 @@ class DefaultSessionOrchestrator(
         graph: WorkflowGraph,
         config: OrchestrationConfig,
     ): WorkflowResult {
-        val effectives = effectivesFor(config)
         val stageId = orchestrationRepository.getState(sessionId).currentStageId
             ?: return WorkflowResult.Failed(sessionId, "resume: no currentStageId", retryExhausted = false)
         log.info("[Orchestrator] resuming session={} workflow={} stage={}", sessionId.value, graph.id, stageId.value)
         val base = ExecutionContext(graph, sessionId, 0, stageId, config, null, null)
         val enriched = base.enrich()
-        return when (val result = enterStage(enriched, stageId, effectives)) {
-            is StepResult.Continue -> step(result.ctx.copy(currentStageId = stageId), effectives)
+        return when (val result = enterStage(enriched, stageId)) {
+            is StepResult.Continue -> step(result.ctx.copy(currentStageId = stageId))
             is StepResult.Terminal -> result.result
         }
     }
@@ -200,7 +207,6 @@ class DefaultSessionOrchestrator(
     private suspend fun executeMove(
         ctx: EnrichedExecutionContext,
         decision: TransitionDecision.Move,
-        effectives: RunEffectives,
     ): StepResult {
         val nextStageId = decision.to
 
@@ -236,7 +242,7 @@ class DefaultSessionOrchestrator(
         // events (InferenceStarted, artifacts, …) are recorded ahead of the TransitionExecuted
         // that marks entering it, so the log shows the stage running before it was entered.
         val advancedTo = advanceStage(ctx.sessionId, ctx.currentStageId, decision)
-        return when (val result = enterStage(ctx.copy(currentStageId = advancedTo), nextStageId, effectives)) {
+        return when (val result = enterStage(ctx.copy(currentStageId = advancedTo), nextStageId)) {
             is StepResult.Continue -> StepResult.Continue(result.ctx)
             is StepResult.Terminal -> result
         }
@@ -246,14 +252,16 @@ class DefaultSessionOrchestrator(
     private suspend fun enterStage(
         ctx: EnrichedExecutionContext,
         stageId: StageId,
-        effectives: RunEffectives,
     ): StepResult {
         log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}")
         while (true) {
             if (isCancelled(ctx.sessionId)) {
                 return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId))
             }
-            when (val result = executeStage(ctx.sessionId, stageId, ctx.graph, ctx.session, ctx.config, effectives)) {
+            val result = subagentRunner.run(
+                SubagentRunRequest(ctx.sessionId, stageId, ctx.graph, ctx.session, ctx.config),
+            ).outcome
+            when (result) {
                 is StageExecutionResult.Success ->
                     return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1))
 

core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt

@@ -15,6 +15,8 @@ import com.correx.core.inference.GenerationConfig
 import com.correx.core.inference.InferenceRequest
 import com.correx.core.inference.ResponseFormat
 import com.correx.core.kernel.execution.ReplayStrategy
+import com.correx.core.kernel.orchestration.subagent.InSessionSubagentRunner
+import com.correx.core.kernel.orchestration.subagent.SubagentRunner
 import com.correx.core.kernel.execution.WorkflowResult
 import com.correx.core.kernel.replay.ReplayInferenceProvider
 import com.correx.core.kernel.retry.exception.ReplayArtifactMissingException
@@ -67,6 +69,12 @@ class ReplayOrchestrator(
 ) {
     private val replayProvider = ReplayInferenceProvider(repositories.eventStore)
 
+    override val subagentRunner: SubagentRunner = InSessionSubagentRunner(
+        executeStage = { sid, stg, graph, session, cfg ->
+            executeStage(sid, stg, graph, session, cfg, effectivesFor(cfg))
+        },
+    )
+
     override suspend fun run(
         sessionId: SessionId,
         graph: WorkflowGraph,

core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt

@@ -110,6 +110,7 @@ import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonObject
 import com.correx.core.journal.DecisionJournalRenderer
 import com.correx.core.journal.DefaultDecisionJournalRepository
+import com.correx.core.kernel.orchestration.subagent.SubagentRunner
 import org.slf4j.LoggerFactory
 import java.util.*
 import java.util.concurrent.*
@@ -156,6 +157,8 @@ abstract class SessionOrchestrator(
     private val approvalEngine: ApprovalEngine = engines.approvalEngine
     private val approvalRepository: DefaultApprovalRepository = repositories.approvalRepository
     internal abstract val cancellations: ConcurrentHashMap<SessionId, AtomicBoolean>
+
+    protected abstract val subagentRunner: SubagentRunner
     internal val pendingApprovals: ConcurrentHashMap<ApprovalRequestId, CompletableDeferred<ApprovalDecision>> =
         ConcurrentHashMap()
 
@@ -308,7 +311,8 @@ abstract class SessionOrchestrator(
         val repoMapEntries = buildRepoMapEntries(sessionId)
         val needsEntries = buildNeedsArtifactEntries(sessionId, stageConfig)
         var accumulatedEntries =
-            systemPrompt + journalEntries + repoMapEntries + needsEntries + schemaEntries + promptEntries + steeringEntries
+            systemPrompt + journalEntries + repoMapEntries + needsEntries +
+                schemaEntries + promptEntries + steeringEntries
         val contextPack = contextPackBuilder.build(
             id = ContextPackId(UUID.randomUUID().toString()),
             sessionId = sessionId,

core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/subagent/InSessionSubagentRunner.kt

@@ -0,0 +1,25 @@
+package com.correx.core.kernel.orchestration.subagent
+
+import com.correx.core.events.types.SessionId
+import com.correx.core.events.types.StageId
+import com.correx.core.kernel.orchestration.OrchestrationConfig
+import com.correx.core.sessions.Session
+import com.correx.core.transitions.execution.StageExecutionResult
+import com.correx.core.transitions.graph.WorkflowGraph
+
+/**
+ * Runs the stage in the parent session by delegating to the orchestrator's existing
+ * stage-execution function. No new isolation; reuses journal + CAS + validation.
+ */
+class InSessionSubagentRunner(
+    private val executeStage: suspend (
+        SessionId, StageId, WorkflowGraph, Session, OrchestrationConfig,
+    ) -> StageExecutionResult,
+) : SubagentRunner {
+    override suspend fun run(request: SubagentRunRequest): SubagentRunResult =
+        SubagentRunResult(
+            executeStage(
+                request.sessionId, request.stageId, request.graph, request.session, request.config,
+            ),
+        )
+}

core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/subagent/SubagentRunner.kt

@@ -0,0 +1,29 @@
+package com.correx.core.kernel.orchestration.subagent
+
+import com.correx.core.events.types.SessionId
+import com.correx.core.events.types.StageId
+import com.correx.core.kernel.orchestration.OrchestrationConfig
+import com.correx.core.sessions.Session
+import com.correx.core.transitions.execution.StageExecutionResult
+import com.correx.core.transitions.graph.WorkflowGraph
+
+/** One sequential one-shot subagent run for a single plan/workflow stage. */
+data class SubagentRunRequest(
+    val sessionId: SessionId,
+    val stageId: StageId,
+    val graph: WorkflowGraph,
+    val session: Session,
+    val config: OrchestrationConfig,
+)
+
+data class SubagentRunResult(val outcome: StageExecutionResult)
+
+/**
+ * Seam between the execution loop and "run one stage to completion." The default
+ * [InSessionSubagentRunner] runs the stage in the parent session. A future
+ * child-session runner can implement this interface with no caller changes.
+ * Sequential by contract — callers must not invoke concurrently for one session.
+ */
+fun interface SubagentRunner {
+    suspend fun run(request: SubagentRunRequest): SubagentRunResult
+}

testing/integration/src/test/kotlin/SubagentRunnerSeamTest.kt

@@ -0,0 +1,140 @@
+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.testing.fixtures.context.ContextFixtures
+import com.correx.core.events.events.TransitionExecutedEvent
+import com.correx.core.events.events.WorkflowCompletedEvent
+import com.correx.core.events.execution.RetryPolicy
+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.InferenceState
+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.InferenceFixtures
+import com.correx.testing.fixtures.cyclePolicyMissingValidator
+import com.correx.testing.fixtures.transitions.TransitionFixtures
+import com.correx.testing.kernel.MockSessionEventReplayer
+import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Test
+
+/**
+ * Regression guard for the [com.correx.core.kernel.orchestration.subagent.SubagentRunner] seam:
+ * stage execution is now dispatched through the runner rather than calling executeStage inline.
+ * The default in-session runner must produce behaviour identical to the prior path — a workflow
+ * still runs every non-terminal stage to completion. (executeStage is module-internal, so the
+ * runner cannot be wrapped from this test module; we assert per-stage execution end-to-end.)
+ */
+class SubagentRunnerSeamTest {
+
+    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 val orchestrator = DefaultSessionOrchestrator(
+        repositories = repositories,
+        engines = OrchestratorEngines(
+            transitionResolver = TransitionFixtures.simpleResolver(),
+            contextPackBuilder = ContextFixtures.simpleBuilder(),
+            inferenceRouter = InferenceFixtures.fixedRouter(),
+            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),
+    )
+
+    @Test
+    fun `workflow runs every non-terminal stage through the runner and completes`(): Unit = runBlocking {
+        val sessionId = SessionId("seam-test-1")
+        val stageA = StageId("A")
+        val stageB = StageId("B")
+        val done = StageId("done")
+
+        val graph = WorkflowGraph(
+            id = "seam-test",
+            stages = mapOf(
+                stageA to StageConfig(allowedTools = emptySet()),
+                stageB to StageConfig(allowedTools = emptySet()),
+                done to StageConfig(),
+            ),
+            transitions = setOf(
+                TransitionEdge(id = TransitionId("t1"), from = stageA, to = stageB, condition = { true }),
+                TransitionEdge(id = TransitionId("t2"), from = stageB, to = done, condition = { true }),
+            ),
+            start = stageA,
+        )
+
+        orchestrator.run(sessionId, graph, defaultConfig)
+
+        val events = eventStore.read(sessionId)
+        assertNotNull(
+            events.find { it.payload is WorkflowCompletedEvent },
+            "workflow should complete through the subagent runner seam",
+        )
+        // Stage A is the start (entered, no transition-to event); B is entered via the runner
+        // after the A->B transition; `done` is a terminal sentinel (no TransitionExecuted).
+        val transitions = events.mapNotNull { it.payload as? TransitionExecutedEvent }
+        assertEquals(
+            listOf(stageB),
+            transitions.map { it.to },
+            "stage B should be entered via the runner after stage A completes",
+        )
+    }
+}