fix(sessions): terminalize session on workflow completion

DefaultSessionReducer skipped WorkflowCompletedEvent to avoid terminalizing on freestyle planning's mid-session completion, but that left execution/normal completions stuck ACTIVE forever — headless watchers and approval loops never saw a terminal state. Add workflowId to WorkflowCompletedEvent (defaulted for replay compat), pass graph.id at all completeWorkflow sites, and terminalize to COMPLETED unless it's the freestyle_planning handoff. Correct stale tests that pinned the ACTIVE-forever behavior.
This commit is contained in:
2026-07-02 12:23:39 +04:00
parent c1de2281c8
commit 1a9e1019ec
10 changed files with 58 additions and 16 deletions
@@ -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
@@ -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),
)
}
@@ -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)
}
@@ -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)
}
@@ -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
@@ -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(