diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index b3dd4c61..50a6c56e 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/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 = ConcurrentHashMap() + 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)) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt index 3e1eb012..ec37f21d 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt +++ b/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, 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 5b2afd9d..feb35dca 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 @@ -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 + + protected abstract val subagentRunner: SubagentRunner internal val pendingApprovals: ConcurrentHashMap> = 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, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/subagent/InSessionSubagentRunner.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/subagent/InSessionSubagentRunner.kt new file mode 100644 index 00000000..291b75d8 --- /dev/null +++ b/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, + ), + ) +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/subagent/SubagentRunner.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/subagent/SubagentRunner.kt new file mode 100644 index 00000000..eea929f3 --- /dev/null +++ b/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 +} diff --git a/testing/integration/src/test/kotlin/SubagentRunnerSeamTest.kt b/testing/integration/src/test/kotlin/SubagentRunnerSeamTest.kt new file mode 100644 index 00000000..ebad05b6 --- /dev/null +++ b/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 { + 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", + ) + } +}