diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index 22f3048b..9bfb6744 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -34,6 +34,8 @@ import com.correx.core.sessions.SessionSummaryProjector import com.correx.core.tools.registry.ToolRegistry import com.correx.core.events.orchestration.OrchestrationStatus import kotlinx.datetime.Clock +import kotlinx.datetime.DateTimeUnit +import kotlinx.datetime.minus import java.util.concurrent.ConcurrentHashMap import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers @@ -348,6 +350,12 @@ class ServerModule( } private fun resumeAbandonedSessions() { + val maxAgeMinutes = configHolder.get().orchestration.resumeAbandonedMaxAgeMinutes + if (maxAgeMinutes <= 0) { + log.info("resumeAbandoned: disabled (resume_abandoned_max_age_minutes={})", maxAgeMinutes) + return + } + val cutoff = Clock.System.now().minus(maxAgeMinutes, DateTimeUnit.MINUTE) eventStore.allSessionIds().forEach { sessionId -> val orchState = orchestrationRepository.getState(sessionId) // Skip PAUSED sessions — they are waiting for user approval, not for computation. @@ -355,6 +363,14 @@ class ServerModule( // approves, submitApprovalDecision emits OrchestrationResumedEvent which triggers // the subscription above to continue the session. if (orchState.status != OrchestrationStatus.RUNNING) return@forEach + val lastEventAt = eventStore.read(sessionId).lastOrNull()?.metadata?.timestamp ?: return@forEach + if (lastEventAt < cutoff) { + log.info( + "resumeAbandoned: skipping stale session={} (last event {}, cutoff {}m) — POST /resume revives it", + sessionId.value, lastEventAt, maxAgeMinutes, + ) + return@forEach + } val graph = workflowRegistry.find(orchState.workflowId) ?: run { log.warn("resumeAbandoned: no graph for session={} workflowId={}", sessionId.value, orchState.workflowId) return@forEach diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt index eab9e4df..e34640b4 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt @@ -10,6 +10,9 @@ import com.correx.core.events.events.ApprovalDecisionResolvedEvent import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.ArtifactContentStoredEvent import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.ArtifactValidatingEvent +import com.correx.core.events.events.ExecutionPlanLockedEvent import com.correx.core.events.events.ChatSessionStartedEvent import com.correx.core.events.events.ChatTurnEvent import com.correx.core.events.events.InferenceCompletedEvent @@ -243,6 +246,23 @@ suspend fun domainEventToServerMessage( sequence = seq, sessionSequence = sessionSequence, ) + is ArtifactValidatedEvent -> ServerMessage.ArtifactValidated( + sessionId = p.sessionId, + stageId = p.stageId, + artifactId = p.artifactId, + sequence = seq, + sessionSequence = sessionSequence, + ) + // Transient pre-validation marker, emitted microseconds before Validated or a + // stage failure — either of those carries the outcome the operator cares about. + is ArtifactValidatingEvent -> null + is ExecutionPlanLockedEvent -> ServerMessage.PlanLocked( + sessionId = p.sessionId, + workflowId = p.workflowId, + stageIds = p.stageIds, + sequence = seq, + sessionSequence = sessionSequence, + ) is ModelLoadedEvent -> ServerMessage.ModelChanged( modelId = p.modelId, providerId = p.providerId.value, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt b/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt index 93a46cf6..e762c82e 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt @@ -30,12 +30,7 @@ class NarrationSubscriber( ) { private data class SessionLane(val channel: Channel, var used: Int) - // Latest approval awaiting a decision per session, so a pause narration can name the - // tool and show the proposed change instead of the bare "APPROVAL_PENDING" reason. - private data class PendingApproval(val toolName: String, val tier: String, val preview: String) - private val lanes = ConcurrentHashMap() - private val pendingApprovals = ConcurrentHashMap() fun start() { eventStore.subscribeAll().onEach { handle(it) }.launchIn(scope) @@ -45,51 +40,84 @@ class NarrationSubscriber( val sid = event.metadata.sessionId when (val p = event.payload) { is WorkflowStartedEvent -> lanes[sid.value]?.let { it.used = 0 } - is ApprovalRequestedEvent -> pendingApprovals[sid.value] = PendingApproval( - toolName = p.toolName ?: "a tool", - tier = p.tier.toString(), - preview = p.preview.orEmpty(), + // Approval pauses narrate off this event, not OrchestrationPaused: it carries the + // tool/tier/preview facts directly, where the paused event forced a racy lookup + // against state the next event had not yet written. + is ApprovalRequestedEvent -> enqueue( + sid, + NarrationTrigger( + kind = "paused", + instruction = approvalInstruction(p), + stageId = p.stageId?.value, + ), ) is StageCompletedEvent -> enqueue( sid, - NarrationTrigger(kind = "stage_completed", instruction = "Stage ${p.stageId.value} completed. Summarise what was accomplished."), + NarrationTrigger( + kind = "stage_completed", + instruction = "Stage ${p.stageId.value} completed. Summarise what was accomplished.", + stageId = p.stageId.value, + ), ) is StageFailedEvent -> enqueue( sid, - NarrationTrigger(kind = "stage_failed", instruction = "Stage ${p.stageId.value} failed: ${p.reason}. Explain what went wrong."), + NarrationTrigger( + kind = "stage_failed", + instruction = "Stage ${p.stageId.value} failed: ${p.reason}. Explain what went wrong.", + stageId = p.stageId.value, + ), ) is WorkflowCompletedEvent -> enqueue( sid, - NarrationTrigger(kind = "workflow_completed", instruction = "The workflow finished successfully. Provide a brief summary."), + NarrationTrigger( + kind = "workflow_completed", + instruction = "The workflow finished successfully. Provide a brief summary.", + stageId = p.terminalStageId.value, + ), ) is WorkflowFailedEvent -> enqueue( sid, - NarrationTrigger(kind = "workflow_failed", instruction = "The workflow failed: ${p.reason}. Explain the failure to the user."), + NarrationTrigger( + kind = "workflow_failed", + instruction = "The workflow failed in stage ${p.stageId.value}: ${p.reason}. Explain the failure to the user.", + stageId = p.stageId.value, + ), ) is ExecutionPlanRejectedEvent -> enqueue( sid, - NarrationTrigger(kind = "plan_rejected", instruction = "The execution plan was rejected (${p.source}): ${p.reason}. Explain to the user what happened and what they can do next."), - ) - is OrchestrationPausedEvent -> enqueue( - sid, - NarrationTrigger(kind = "paused", instruction = pauseInstruction(sid, p)), + NarrationTrigger( + kind = "plan_rejected", + instruction = "The execution plan was rejected (${p.source}): ${p.reason}. Explain to the user what happened and what they can do next.", + ), ) + is OrchestrationPausedEvent -> { + // Approval pauses are narrated from ApprovalRequestedEvent above. + if (!p.reason.contains("APPROVAL", ignoreCase = true)) { + enqueue( + sid, + NarrationTrigger( + kind = "paused", + instruction = "Orchestration paused in stage ${p.stageId.value}: ${p.reason}. Inform the user.", + stageId = p.stageId.value, + ), + ) + } + } else -> Unit } } - private fun pauseInstruction(sessionId: SessionId, event: OrchestrationPausedEvent): String { - val pending = pendingApprovals[sessionId.value] - if (!event.reason.contains("APPROVAL", ignoreCase = true) || pending == null) { - return "Orchestration paused in stage ${event.stageId.value}: ${event.reason}. Inform the user." - } + private fun approvalInstruction(event: ApprovalRequestedEvent): String { + val stage = event.stageId?.value ?: "the current stage" + val subject = event.toolName?.let { "of the '$it' tool (tier ${event.tier})" } + ?: "(tier ${event.tier})" return buildString { - append("The workflow paused in stage ${event.stageId.value} awaiting your approval of the ") - append("'${pending.toolName}' tool (tier ${pending.tier}). ") - append("Tell the operator what it is about to do and why it needs sign-off.") - if (pending.preview.isNotBlank()) { + append("The workflow paused in stage $stage awaiting your approval $subject. ") + append("Tell the operator what is proposed and why it needs sign-off.") + val preview = event.preview.orEmpty() + if (preview.isNotBlank()) { append("\n\nProposed change:\n") - append(pending.preview.take(MAX_PREVIEW_CHARS)) + append(preview.take(MAX_PREVIEW_CHARS)) } } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index fa968451..3d0f5197 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -391,6 +391,27 @@ sealed interface ServerMessage { override val sessionSequence: Long, ) : ServerMessage, SessionMessage + @Serializable + @SerialName("artifact.validated") + data class ArtifactValidated( + val sessionId: SessionId, + val stageId: StageId, + val artifactId: ArtifactId, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + + /** A freestyle execution plan was locked; phase 2 runs [stageIds] as workflow [workflowId]. */ + @Serializable + @SerialName("plan.locked") + data class PlanLocked( + val sessionId: SessionId, + val workflowId: String, + val stageIds: List, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + /** * Full artifact listing for a session, returned in response to a ListArtifacts request. * Not event-derived (it is a snapshot read), so cursors are null. diff --git a/apps/tui-go/internal/app/server.go b/apps/tui-go/internal/app/server.go index b732d2cd..8030fd10 100644 --- a/apps/tui-go/internal/app/server.go +++ b/apps/tui-go/internal/app/server.go @@ -227,6 +227,15 @@ func (m *Model) applyServer(msg protocol.ServerMessage) { if s := m.session(msg.SessionID); s != nil { s.addEvent(nowMillis(), "ArtifactCreated", msg.ArtifactID) } + case protocol.TypeArtifactValid: + if s := m.session(msg.SessionID); s != nil { + s.addEvent(nowMillis(), "ArtifactValidated", msg.ArtifactID) + } + case protocol.TypePlanLocked: + if s := m.session(msg.SessionID); s != nil { + s.addEvent(nowMillis(), "PlanLocked", strings.Join(msg.StageIDs, " → ")) + s.LastOutput = "plan locked: " + strings.Join(msg.StageIDs, " → ") + } case protocol.TypeWorkspaceBound: if s := m.session(msg.SessionID); s != nil { s.WorkspaceRoot = msg.WorkspaceRoot diff --git a/apps/tui-go/internal/protocol/protocol.go b/apps/tui-go/internal/protocol/protocol.go index d661af62..5a05cf89 100644 --- a/apps/tui-go/internal/protocol/protocol.go +++ b/apps/tui-go/internal/protocol/protocol.go @@ -39,6 +39,8 @@ const ( TypeProviderStatus = "provider.status_changed" TypeProtocolError = "protocol_error" TypeArtifactCreated = "artifact.created" + TypeArtifactValid = "artifact.validated" + TypePlanLocked = "plan.locked" TypeRouterResponse = "router.response" TypeSnapshotComplete = "snapshot_complete" TypeWorkflowList = "workflow.list" @@ -81,6 +83,9 @@ type ServerMessage struct { Outcome string `json:"outcome"` // approval.resolved: APPROVED | REJECTED | AUTO_APPROVED WorkspaceRoot string `json:"workspaceRoot"` // session.workspace_bound: the bound cwd + // plan.locked: the compiled phase-2 stage ids in execution order + StageIDs []string `json:"stageIds"` + // inference.retry AttemptNumber int `json:"attemptNumber"` MaxAttempts int `json:"maxAttempts"` @@ -224,6 +229,7 @@ func (m ServerMessage) IsEventBearing() bool { TypeToolStarted, TypeToolCompleted, TypeToolFailed, TypeToolRejected, TypeToolAssessed, TypeApprovalRequired, TypeApprovalResolved, TypeWorkspaceBound, TypeArtifactCreated, + TypeArtifactValid, TypePlanLocked, TypeRouterNarration: return true default: diff --git a/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt b/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt index 9a81338d..00352588 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt @@ -583,6 +583,8 @@ object ConfigLoader { stageTimeoutMs = asLong(orchestrationSection["stage_timeout_ms"], 180_000), journalCompactionTokenThreshold = asInt(orchestrationSection["journal_compaction_token_threshold"], 2_000), + resumeAbandonedMaxAgeMinutes = + asLong(orchestrationSection["resume_abandoned_max_age_minutes"], 1_440), ) val modelsSettings = ModelsSettings( diff --git a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt index 04dcd8b5..6eeb01d3 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt @@ -28,6 +28,12 @@ data class CorrexConfig( data class OrchestrationKnobs( val stageTimeoutMs: Long = 180_000, val journalCompactionTokenThreshold: Int = 2_000, + /** + * Boot-time abandoned-session pickup only resumes sessions whose last event is at most + * this old; anything staler stays parked (POST /sessions/{id}/resume revives it on demand). + * 0 disables auto-resume entirely. + */ + val resumeAbandonedMaxAgeMinutes: Long = 1_440, ) @Serializable diff --git a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfigWriter.kt b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfigWriter.kt index 6f36b59a..b09b36c0 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfigWriter.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfigWriter.kt @@ -84,6 +84,7 @@ object CorrexConfigWriter { b.section("orchestration") b.kv("stage_timeout_ms", cfg.orchestration.stageTimeoutMs) b.kv("journal_compaction_token_threshold", cfg.orchestration.journalCompactionTokenThreshold) + b.kv("resume_abandoned_max_age_minutes", cfg.orchestration.resumeAbandonedMaxAgeMinutes) b.section("personalization") b.kv("enabled", cfg.personalization.enabled) diff --git a/core/config/src/main/kotlin/com/correx/core/config/EditableConfig.kt b/core/config/src/main/kotlin/com/correx/core/config/EditableConfig.kt index be59ff47..cd9b0676 100644 --- a/core/config/src/main/kotlin/com/correx/core/config/EditableConfig.kt +++ b/core/config/src/main/kotlin/com/correx/core/config/EditableConfig.kt @@ -175,6 +175,13 @@ object EditableConfig { getString = { it.orchestration.journalCompactionTokenThreshold.toString() }, withString = { c, v -> orc(c) { it.copy(journalCompactionTokenThreshold = int(JCT_KEY, v)) } }, ), + EditableField( + "orchestration.resume_abandoned_max_age_minutes", ConfigFieldType.LONG, + getString = { it.orchestration.resumeAbandonedMaxAgeMinutes.toString() }, + withString = { c, v -> + orc(c) { it.copy(resumeAbandonedMaxAgeMinutes = lng("orchestration.resume_abandoned_max_age_minutes", v)) } + }, + ), ) private const val JCT_KEY = "orchestration.journal_compaction_token_threshold" diff --git a/core/router/src/main/kotlin/com/correx/core/router/RouterContextBuilder.kt b/core/router/src/main/kotlin/com/correx/core/router/RouterContextBuilder.kt index 0e86e9ca..d694ec1b 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/RouterContextBuilder.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/RouterContextBuilder.kt @@ -281,9 +281,9 @@ class DefaultRouterContextBuilder( ) } - private fun buildWorkflowStatusContent(state: RouterState): String { + private fun buildWorkflowStatusContent(state: RouterState, stageOverride: String? = null): String { val status = capitalizeFirst(state.workflowStatus.name.lowercase()) - val stage = state.currentStageId?.value ?: "none" + val stage = stageOverride ?: state.currentStageId?.value ?: "none" return "Status: $status, Stage: $stage" } @@ -369,8 +369,8 @@ class DefaultRouterContextBuilder( ) val workflowStatusEntry = buildContextEntry( sourceType = "workflowStatus", - sourceId = state.currentStageId?.value ?: "none", - content = buildWorkflowStatusContent(state), + sourceId = trigger.stageId ?: state.currentStageId?.value ?: "none", + content = buildWorkflowStatusContent(state, trigger.stageId), layer = ContextLayer.L0, role = EntryRole.SYSTEM, ) diff --git a/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt b/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt index 2f7b1116..52a24408 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt @@ -15,6 +15,7 @@ import com.correx.core.events.types.InferenceRequestId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.inference.Embedder +import com.correx.core.inference.FinishReason import com.correx.core.inference.InferenceRequest import com.correx.core.inference.InferenceRouter import com.correx.core.inference.ModelCapability @@ -153,7 +154,18 @@ class DefaultRouterFacade( responseFormat = ResponseFormat.Text, ) val inferenceResponse = provider.infer(inferenceRequest) - val content = inferenceResponse.text + val rawContent = inferenceResponse.text + // A length-finish with no visible text means the model burned the whole completion + // budget on hidden reasoning — surface that instead of rendering a silent blank turn. + val content = rawContent.ifBlank { + if (inferenceResponse.finishReason is FinishReason.Length) { + "The model used its entire ${config.generationConfig.maxTokens}-token response budget " + + "without producing an answer (likely spent on hidden reasoning). " + + "Raise [router.generation] max_tokens or lower the model's reasoning budget, then ask again." + } else { + "The model returned an empty response (finish reason: ${inferenceResponse.finishReason::class.simpleName}). Try again." + } + } // Emit ROUTER turn event emitChatTurn( @@ -162,19 +174,20 @@ class DefaultRouterFacade( tokensUsed = inferenceResponse.tokensUsed, ) - if (mode == ChatMode.STEERING) { + val steeringEmitted = mode == ChatMode.STEERING && rawContent.isNotBlank() + if (steeringEmitted) { val validationError = validateSteering?.invoke(content) if (validationError == null) { emitSteeringNote(sessionId, content, effectiveStageId) } } - return RouterResponse(content = content, steeringEmitted = (mode == ChatMode.STEERING)) + return RouterResponse(content = content, steeringEmitted = steeringEmitted) } override suspend fun narrate(sessionId: SessionId, trigger: NarrationTrigger) { val state = routerRepository.getRouterState(sessionId) - val effectiveStageId = state.currentStageId ?: StageId.NONE + val effectiveStageId = trigger.stageId?.let { StageId(it) } ?: state.currentStageId ?: StageId.NONE val contextPack = routerContextBuilder.buildNarrationContext(state, trigger, config.tokenBudget) diff --git a/core/router/src/main/kotlin/com/correx/core/router/model/NarrationTrigger.kt b/core/router/src/main/kotlin/com/correx/core/router/model/NarrationTrigger.kt index e3e150a3..f4ab1b38 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/model/NarrationTrigger.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/model/NarrationTrigger.kt @@ -11,4 +11,10 @@ data class NarrationTrigger( val kind: String, /** Human-readable instruction injected as the final context entry for this narration. */ val instruction: String, + /** + * Stage the triggering event belongs to. Carried on the trigger because the router + * projection's currentStageId is already cleared (workflow terminal events) or not yet + * written when the narrator fires — reading it back from state races the reducer. + */ + val stageId: String? = null, )