diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt index e6f80021..a4168ccd 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt @@ -21,6 +21,10 @@ data class WorkflowCompletedEvent( val sessionId: SessionId, val terminalStageId: StageId, val totalStages: Int, + // The workflow that completed. Lets the session reducer tell a session-terminal completion from + // the freestyle planning workflow, which completes mid-session before handing off to execution. + // Defaulted for backward-compatible replay of events stored before this field existed. + val workflowId: String = "", ) : EventPayload @Serializable 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 97a2678f..18fd297c 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 @@ -168,7 +168,7 @@ class DefaultSessionOrchestrator( } is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) { - completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount) + completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount, enriched.graph.id) } else { // No outgoing edge matched: the stage finished but produced nothing that satisfies a // transition (e.g. the analyst never emitted a validated `analysis`). Treat it as a @@ -335,7 +335,7 @@ class DefaultSessionOrchestrator( if (isTerminal(ctx.graph, nextStageId)) { // Terminal stage is a sentinel — not executed, so don't count it. return StepResult.Terminal( - completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount), + completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id), ) } 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 477011d9..1907aa39 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 @@ -117,7 +117,7 @@ class ReplayOrchestrator( val nextStageId = decision.to if (isTerminal(ctx.graph, nextStageId)) { - return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount) + return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id) } when (val result = enterStage(ctx, nextStageId, session)) { @@ -131,7 +131,7 @@ class ReplayOrchestrator( } is TransitionDecision.Stay -> if (isTerminal(ctx.graph, ctx.currentStageId)) { - completeWorkflow(ctx.sessionId, ctx.currentStageId, ctx.stageCount) + completeWorkflow(ctx.sessionId, ctx.currentStageId, ctx.stageCount, ctx.graph.id) } else { step(ctx) } 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 6f4691d3..3f301524 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 @@ -2065,13 +2065,14 @@ abstract class SessionOrchestrator( sessionId: SessionId, terminalStageId: StageId, stageCount: Int, + workflowId: String, ): WorkflowResult.Completed { log.debug( "[Orchestrator] COMPLETED session={} terminalStage={} stages={}", sessionId.value, terminalStageId.value, stageCount, ) correlateCritiqueOutcomes(sessionId) - emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount)) + emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId)) cancellations.remove(sessionId) return WorkflowResult.Completed(sessionId, terminalStageId) } diff --git a/core/sessions/src/main/kotlin/com/correx/core/sessions/DefaultSessionReducer.kt b/core/sessions/src/main/kotlin/com/correx/core/sessions/DefaultSessionReducer.kt index d196721e..07d76646 100644 --- a/core/sessions/src/main/kotlin/com/correx/core/sessions/DefaultSessionReducer.kt +++ b/core/sessions/src/main/kotlin/com/correx/core/sessions/DefaultSessionReducer.kt @@ -8,8 +8,13 @@ import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowFailedEvent +// The freestyle two-phase planning workflow. Its completion hands off to the execution workflow +// rather than ending the session, so it must not terminalize (see reduce()). +private const val PLANNING_WORKFLOW_ID = "freestyle_planning" + class DefaultSessionReducer : SessionReducer { override fun reduce( @@ -24,13 +29,19 @@ class DefaultSessionReducer : SessionReducer { SessionStatus.FAILED // A failed workflow is terminal for the session: no current workflow recovers after - // one fails (freestyle's planning→execution handoff only fires on WorkflowCompleted). - // Without this, an exhausted-retry WorkflowFailed left the session ACTIVE forever, so - // clients/approval loops never saw a terminal state. WorkflowCompleted is deliberately - // NOT mapped here — planning emits its own mid-session, so it isn't session-terminal. + // one fails. is WorkflowFailedEvent -> SessionStatus.FAILED + // A completed workflow is session-terminal EXCEPT the freestyle planning workflow, which + // completes mid-session and hands off to the execution workflow (the handoff fires on + // planning's WorkflowCompleted — see ServerModule). Terminalizing on it would falsely mark + // the session done during the planning→execution gap; leaving it ACTIVE lets the real + // (execution) completion terminalize. Without this, a normal single-workflow session — and + // the execution phase — stayed ACTIVE forever, so clients/approval loops never saw done. + is WorkflowCompletedEvent -> + if (payload.workflowId == PLANNING_WORKFLOW_ID) state.status else SessionStatus.COMPLETED + is StageCompletedEvent, is TransitionExecutedEvent -> SessionStatus.ACTIVE diff --git a/core/sessions/src/test/kotlin/com/correx/core/sessions/DefaultSessionReducerTest.kt b/core/sessions/src/test/kotlin/com/correx/core/sessions/DefaultSessionReducerTest.kt index c47ecaf0..d1d2d337 100644 --- a/core/sessions/src/test/kotlin/com/correx/core/sessions/DefaultSessionReducerTest.kt +++ b/core/sessions/src/test/kotlin/com/correx/core/sessions/DefaultSessionReducerTest.kt @@ -70,6 +70,32 @@ class DefaultSessionReducerTest { assertEquals(SessionStatus.FAILED, reducer.reduce(initialState, event).status) } + @Test + fun `WorkflowCompletedEvent flips session status to COMPLETED`() { + val event = stored( + com.correx.core.events.events.WorkflowCompletedEvent( + sessionId = sessionId, + terminalStageId = com.correx.core.events.types.StageId("report"), + totalStages = 3, + workflowId = "review_loop", + ), + ) + assertEquals(SessionStatus.COMPLETED, reducer.reduce(initialState, event).status) + } + + @Test + fun `WorkflowCompletedEvent for freestyle planning stays ACTIVE (hands off to execution)`() { + val event = stored( + com.correx.core.events.events.WorkflowCompletedEvent( + sessionId = sessionId, + terminalStageId = com.correx.core.events.types.StageId("architect"), + totalStages = 2, + workflowId = "freestyle_planning", + ), + ) + assertEquals(SessionStatus.ACTIVE, reducer.reduce(initialState, event).status) + } + @Test fun `unrelated event leaves boundProfile unchanged`() { val state = initialState.copy( diff --git a/testing/projections/src/test/kotlin/DefaultSessionReducerTest.kt b/testing/projections/src/test/kotlin/DefaultSessionReducerTest.kt index 0b4ac707..872f8697 100644 --- a/testing/projections/src/test/kotlin/DefaultSessionReducerTest.kt +++ b/testing/projections/src/test/kotlin/DefaultSessionReducerTest.kt @@ -75,7 +75,7 @@ class DefaultSessionReducerTest { } @Test - fun `WorkflowCompletedEvent leaves ACTIVE status`() { + fun `WorkflowCompletedEvent flips to COMPLETED status`() { val state = activeState() val result = reducer.reduce( @@ -86,11 +86,11 @@ class DefaultSessionReducerTest { ) ) - assertEquals(SessionStatus.ACTIVE, result.status) + assertEquals(SessionStatus.COMPLETED, result.status) } @Test - fun `WorkflowFailedEvent leaves ACTIVE status`() { + fun `WorkflowFailedEvent flips to FAILED status`() { val state = activeState() val result = reducer.reduce( @@ -101,7 +101,7 @@ class DefaultSessionReducerTest { ) ) - assertEquals(SessionStatus.ACTIVE, result.status) + assertEquals(SessionStatus.FAILED, result.status) } @Test diff --git a/testing/projections/src/test/kotlin/SessionProjectorTest.kt b/testing/projections/src/test/kotlin/SessionProjectorTest.kt index e8ab29ef..808e4107 100644 --- a/testing/projections/src/test/kotlin/SessionProjectorTest.kt +++ b/testing/projections/src/test/kotlin/SessionProjectorTest.kt @@ -35,7 +35,7 @@ class SessionProjectorTest { state = projector.apply(state, it) } - assertEquals(SessionStatus.CREATED, state.status) + assertEquals(SessionStatus.COMPLETED, state.status) } @Test diff --git a/testing/replay/src/test/kotlin/SessionReplayTest.kt b/testing/replay/src/test/kotlin/SessionReplayTest.kt index 58bfa186..e6308a53 100644 --- a/testing/replay/src/test/kotlin/SessionReplayTest.kt +++ b/testing/replay/src/test/kotlin/SessionReplayTest.kt @@ -59,6 +59,6 @@ class SessionReplayTest { val state = replayer.rebuild(sessionId) - assertEquals(SessionStatus.CREATED, state.status) + assertEquals(SessionStatus.COMPLETED, state.status) } } diff --git a/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt b/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt index a87068fc..2b87c0c0 100644 --- a/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt +++ b/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt @@ -80,7 +80,7 @@ class TransitionReplayIntegrationTest { val state = replayer.rebuild(sessionId) assertEquals( - SessionStatus.ACTIVE, + SessionStatus.COMPLETED, state.status ) }