#2: approval pauses narrate from ApprovalRequestedEvent (carries tool/tier/preview/stage directly) instead of OrchestrationPaused + a pending-approvals map the next event had not yet populated — the narrator always lost that race and produced generic text. #3: NarrationTrigger now carries the triggering event's stageId; the narration status line and RouterNarrationEvent.stageId use it instead of the router projection's currentStageId, which terminal events have already cleared ('Stage: none'). #9: ExecutionPlanLocked -> plan.locked and ArtifactValidated -> artifact.validated mapped to the TUI (Go: feed lines, plan shows the compiled stage chain); ArtifactValidating explicitly dropped. #10: blank router turn guard — a length-finish with empty text now emits an explanatory turn naming the max_tokens budget instead of a silent blank; blank steering responses no longer emit steering notes. #12: resumeAbandoned only auto-resumes sessions whose last event is within orchestration.resume_abandoned_max_age_minutes (default 24h, 0 disables); staler sessions stay parked for POST /resume. Knob is TUI-editable and persisted.
This commit is contained in:
@@ -34,6 +34,8 @@ import com.correx.core.sessions.SessionSummaryProjector
|
|||||||
import com.correx.core.tools.registry.ToolRegistry
|
import com.correx.core.tools.registry.ToolRegistry
|
||||||
import com.correx.core.events.orchestration.OrchestrationStatus
|
import com.correx.core.events.orchestration.OrchestrationStatus
|
||||||
import kotlinx.datetime.Clock
|
import kotlinx.datetime.Clock
|
||||||
|
import kotlinx.datetime.DateTimeUnit
|
||||||
|
import kotlinx.datetime.minus
|
||||||
import java.util.concurrent.ConcurrentHashMap
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
import kotlinx.coroutines.CoroutineScope
|
import kotlinx.coroutines.CoroutineScope
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
@@ -348,6 +350,12 @@ class ServerModule(
|
|||||||
}
|
}
|
||||||
|
|
||||||
private fun resumeAbandonedSessions() {
|
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 ->
|
eventStore.allSessionIds().forEach { sessionId ->
|
||||||
val orchState = orchestrationRepository.getState(sessionId)
|
val orchState = orchestrationRepository.getState(sessionId)
|
||||||
// Skip PAUSED sessions — they are waiting for user approval, not for computation.
|
// Skip PAUSED sessions — they are waiting for user approval, not for computation.
|
||||||
@@ -355,6 +363,14 @@ class ServerModule(
|
|||||||
// approves, submitApprovalDecision emits OrchestrationResumedEvent which triggers
|
// approves, submitApprovalDecision emits OrchestrationResumedEvent which triggers
|
||||||
// the subscription above to continue the session.
|
// the subscription above to continue the session.
|
||||||
if (orchState.status != OrchestrationStatus.RUNNING) return@forEach
|
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 {
|
val graph = workflowRegistry.find(orchState.workflowId) ?: run {
|
||||||
log.warn("resumeAbandoned: no graph for session={} workflowId={}", sessionId.value, orchState.workflowId)
|
log.warn("resumeAbandoned: no graph for session={} workflowId={}", sessionId.value, orchState.workflowId)
|
||||||
return@forEach
|
return@forEach
|
||||||
|
|||||||
@@ -10,6 +10,9 @@ import com.correx.core.events.events.ApprovalDecisionResolvedEvent
|
|||||||
import com.correx.core.events.events.ApprovalRequestedEvent
|
import com.correx.core.events.events.ApprovalRequestedEvent
|
||||||
import com.correx.core.events.events.ArtifactContentStoredEvent
|
import com.correx.core.events.events.ArtifactContentStoredEvent
|
||||||
import com.correx.core.events.events.ArtifactCreatedEvent
|
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.ChatSessionStartedEvent
|
||||||
import com.correx.core.events.events.ChatTurnEvent
|
import com.correx.core.events.events.ChatTurnEvent
|
||||||
import com.correx.core.events.events.InferenceCompletedEvent
|
import com.correx.core.events.events.InferenceCompletedEvent
|
||||||
@@ -243,6 +246,23 @@ suspend fun domainEventToServerMessage(
|
|||||||
sequence = seq,
|
sequence = seq,
|
||||||
sessionSequence = sessionSequence,
|
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(
|
is ModelLoadedEvent -> ServerMessage.ModelChanged(
|
||||||
modelId = p.modelId,
|
modelId = p.modelId,
|
||||||
providerId = p.providerId.value,
|
providerId = p.providerId.value,
|
||||||
|
|||||||
+56
-28
@@ -30,12 +30,7 @@ class NarrationSubscriber(
|
|||||||
) {
|
) {
|
||||||
private data class SessionLane(val channel: Channel<NarrationTrigger>, var used: Int)
|
private data class SessionLane(val channel: Channel<NarrationTrigger>, 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<String, SessionLane>()
|
private val lanes = ConcurrentHashMap<String, SessionLane>()
|
||||||
private val pendingApprovals = ConcurrentHashMap<String, PendingApproval>()
|
|
||||||
|
|
||||||
fun start() {
|
fun start() {
|
||||||
eventStore.subscribeAll().onEach { handle(it) }.launchIn(scope)
|
eventStore.subscribeAll().onEach { handle(it) }.launchIn(scope)
|
||||||
@@ -45,51 +40,84 @@ class NarrationSubscriber(
|
|||||||
val sid = event.metadata.sessionId
|
val sid = event.metadata.sessionId
|
||||||
when (val p = event.payload) {
|
when (val p = event.payload) {
|
||||||
is WorkflowStartedEvent -> lanes[sid.value]?.let { it.used = 0 }
|
is WorkflowStartedEvent -> lanes[sid.value]?.let { it.used = 0 }
|
||||||
is ApprovalRequestedEvent -> pendingApprovals[sid.value] = PendingApproval(
|
// Approval pauses narrate off this event, not OrchestrationPaused: it carries the
|
||||||
toolName = p.toolName ?: "a tool",
|
// tool/tier/preview facts directly, where the paused event forced a racy lookup
|
||||||
tier = p.tier.toString(),
|
// against state the next event had not yet written.
|
||||||
preview = p.preview.orEmpty(),
|
is ApprovalRequestedEvent -> enqueue(
|
||||||
|
sid,
|
||||||
|
NarrationTrigger(
|
||||||
|
kind = "paused",
|
||||||
|
instruction = approvalInstruction(p),
|
||||||
|
stageId = p.stageId?.value,
|
||||||
|
),
|
||||||
)
|
)
|
||||||
is StageCompletedEvent -> enqueue(
|
is StageCompletedEvent -> enqueue(
|
||||||
sid,
|
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(
|
is StageFailedEvent -> enqueue(
|
||||||
sid,
|
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(
|
is WorkflowCompletedEvent -> enqueue(
|
||||||
sid,
|
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(
|
is WorkflowFailedEvent -> enqueue(
|
||||||
sid,
|
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(
|
is ExecutionPlanRejectedEvent -> enqueue(
|
||||||
sid,
|
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."),
|
NarrationTrigger(
|
||||||
)
|
kind = "plan_rejected",
|
||||||
is OrchestrationPausedEvent -> enqueue(
|
instruction = "The execution plan was rejected (${p.source}): ${p.reason}. Explain to the user what happened and what they can do next.",
|
||||||
sid,
|
),
|
||||||
NarrationTrigger(kind = "paused", instruction = pauseInstruction(sid, p)),
|
|
||||||
)
|
)
|
||||||
|
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
|
else -> Unit
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private fun pauseInstruction(sessionId: SessionId, event: OrchestrationPausedEvent): String {
|
private fun approvalInstruction(event: ApprovalRequestedEvent): String {
|
||||||
val pending = pendingApprovals[sessionId.value]
|
val stage = event.stageId?.value ?: "the current stage"
|
||||||
if (!event.reason.contains("APPROVAL", ignoreCase = true) || pending == null) {
|
val subject = event.toolName?.let { "of the '$it' tool (tier ${event.tier})" }
|
||||||
return "Orchestration paused in stage ${event.stageId.value}: ${event.reason}. Inform the user."
|
?: "(tier ${event.tier})"
|
||||||
}
|
|
||||||
return buildString {
|
return buildString {
|
||||||
append("The workflow paused in stage ${event.stageId.value} awaiting your approval of the ")
|
append("The workflow paused in stage $stage awaiting your approval $subject. ")
|
||||||
append("'${pending.toolName}' tool (tier ${pending.tier}). ")
|
append("Tell the operator what is proposed and why it needs sign-off.")
|
||||||
append("Tell the operator what it is about to do and why it needs sign-off.")
|
val preview = event.preview.orEmpty()
|
||||||
if (pending.preview.isNotBlank()) {
|
if (preview.isNotBlank()) {
|
||||||
append("\n\nProposed change:\n")
|
append("\n\nProposed change:\n")
|
||||||
append(pending.preview.take(MAX_PREVIEW_CHARS))
|
append(preview.take(MAX_PREVIEW_CHARS))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -391,6 +391,27 @@ sealed interface ServerMessage {
|
|||||||
override val sessionSequence: Long,
|
override val sessionSequence: Long,
|
||||||
) : ServerMessage, SessionMessage
|
) : 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<String>,
|
||||||
|
override val sequence: Long,
|
||||||
|
override val sessionSequence: Long,
|
||||||
|
) : ServerMessage, SessionMessage
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Full artifact listing for a session, returned in response to a ListArtifacts request.
|
* 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.
|
* Not event-derived (it is a snapshot read), so cursors are null.
|
||||||
|
|||||||
@@ -227,6 +227,15 @@ func (m *Model) applyServer(msg protocol.ServerMessage) {
|
|||||||
if s := m.session(msg.SessionID); s != nil {
|
if s := m.session(msg.SessionID); s != nil {
|
||||||
s.addEvent(nowMillis(), "ArtifactCreated", msg.ArtifactID)
|
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:
|
case protocol.TypeWorkspaceBound:
|
||||||
if s := m.session(msg.SessionID); s != nil {
|
if s := m.session(msg.SessionID); s != nil {
|
||||||
s.WorkspaceRoot = msg.WorkspaceRoot
|
s.WorkspaceRoot = msg.WorkspaceRoot
|
||||||
|
|||||||
@@ -39,6 +39,8 @@ const (
|
|||||||
TypeProviderStatus = "provider.status_changed"
|
TypeProviderStatus = "provider.status_changed"
|
||||||
TypeProtocolError = "protocol_error"
|
TypeProtocolError = "protocol_error"
|
||||||
TypeArtifactCreated = "artifact.created"
|
TypeArtifactCreated = "artifact.created"
|
||||||
|
TypeArtifactValid = "artifact.validated"
|
||||||
|
TypePlanLocked = "plan.locked"
|
||||||
TypeRouterResponse = "router.response"
|
TypeRouterResponse = "router.response"
|
||||||
TypeSnapshotComplete = "snapshot_complete"
|
TypeSnapshotComplete = "snapshot_complete"
|
||||||
TypeWorkflowList = "workflow.list"
|
TypeWorkflowList = "workflow.list"
|
||||||
@@ -81,6 +83,9 @@ type ServerMessage struct {
|
|||||||
Outcome string `json:"outcome"` // approval.resolved: APPROVED | REJECTED | AUTO_APPROVED
|
Outcome string `json:"outcome"` // approval.resolved: APPROVED | REJECTED | AUTO_APPROVED
|
||||||
WorkspaceRoot string `json:"workspaceRoot"` // session.workspace_bound: the bound cwd
|
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
|
// inference.retry
|
||||||
AttemptNumber int `json:"attemptNumber"`
|
AttemptNumber int `json:"attemptNumber"`
|
||||||
MaxAttempts int `json:"maxAttempts"`
|
MaxAttempts int `json:"maxAttempts"`
|
||||||
@@ -224,6 +229,7 @@ func (m ServerMessage) IsEventBearing() bool {
|
|||||||
TypeToolStarted, TypeToolCompleted, TypeToolFailed, TypeToolRejected,
|
TypeToolStarted, TypeToolCompleted, TypeToolFailed, TypeToolRejected,
|
||||||
TypeToolAssessed,
|
TypeToolAssessed,
|
||||||
TypeApprovalRequired, TypeApprovalResolved, TypeWorkspaceBound, TypeArtifactCreated,
|
TypeApprovalRequired, TypeApprovalResolved, TypeWorkspaceBound, TypeArtifactCreated,
|
||||||
|
TypeArtifactValid, TypePlanLocked,
|
||||||
TypeRouterNarration:
|
TypeRouterNarration:
|
||||||
return true
|
return true
|
||||||
default:
|
default:
|
||||||
|
|||||||
@@ -583,6 +583,8 @@ object ConfigLoader {
|
|||||||
stageTimeoutMs = asLong(orchestrationSection["stage_timeout_ms"], 180_000),
|
stageTimeoutMs = asLong(orchestrationSection["stage_timeout_ms"], 180_000),
|
||||||
journalCompactionTokenThreshold =
|
journalCompactionTokenThreshold =
|
||||||
asInt(orchestrationSection["journal_compaction_token_threshold"], 2_000),
|
asInt(orchestrationSection["journal_compaction_token_threshold"], 2_000),
|
||||||
|
resumeAbandonedMaxAgeMinutes =
|
||||||
|
asLong(orchestrationSection["resume_abandoned_max_age_minutes"], 1_440),
|
||||||
)
|
)
|
||||||
|
|
||||||
val modelsSettings = ModelsSettings(
|
val modelsSettings = ModelsSettings(
|
||||||
|
|||||||
@@ -28,6 +28,12 @@ data class CorrexConfig(
|
|||||||
data class OrchestrationKnobs(
|
data class OrchestrationKnobs(
|
||||||
val stageTimeoutMs: Long = 180_000,
|
val stageTimeoutMs: Long = 180_000,
|
||||||
val journalCompactionTokenThreshold: Int = 2_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
|
@Serializable
|
||||||
|
|||||||
@@ -84,6 +84,7 @@ object CorrexConfigWriter {
|
|||||||
b.section("orchestration")
|
b.section("orchestration")
|
||||||
b.kv("stage_timeout_ms", cfg.orchestration.stageTimeoutMs)
|
b.kv("stage_timeout_ms", cfg.orchestration.stageTimeoutMs)
|
||||||
b.kv("journal_compaction_token_threshold", cfg.orchestration.journalCompactionTokenThreshold)
|
b.kv("journal_compaction_token_threshold", cfg.orchestration.journalCompactionTokenThreshold)
|
||||||
|
b.kv("resume_abandoned_max_age_minutes", cfg.orchestration.resumeAbandonedMaxAgeMinutes)
|
||||||
|
|
||||||
b.section("personalization")
|
b.section("personalization")
|
||||||
b.kv("enabled", cfg.personalization.enabled)
|
b.kv("enabled", cfg.personalization.enabled)
|
||||||
|
|||||||
@@ -175,6 +175,13 @@ object EditableConfig {
|
|||||||
getString = { it.orchestration.journalCompactionTokenThreshold.toString() },
|
getString = { it.orchestration.journalCompactionTokenThreshold.toString() },
|
||||||
withString = { c, v -> orc(c) { it.copy(journalCompactionTokenThreshold = int(JCT_KEY, v)) } },
|
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"
|
private const val JCT_KEY = "orchestration.journal_compaction_token_threshold"
|
||||||
|
|||||||
@@ -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 status = capitalizeFirst(state.workflowStatus.name.lowercase())
|
||||||
val stage = state.currentStageId?.value ?: "none"
|
val stage = stageOverride ?: state.currentStageId?.value ?: "none"
|
||||||
return "Status: $status, Stage: $stage"
|
return "Status: $status, Stage: $stage"
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -369,8 +369,8 @@ class DefaultRouterContextBuilder(
|
|||||||
)
|
)
|
||||||
val workflowStatusEntry = buildContextEntry(
|
val workflowStatusEntry = buildContextEntry(
|
||||||
sourceType = "workflowStatus",
|
sourceType = "workflowStatus",
|
||||||
sourceId = state.currentStageId?.value ?: "none",
|
sourceId = trigger.stageId ?: state.currentStageId?.value ?: "none",
|
||||||
content = buildWorkflowStatusContent(state),
|
content = buildWorkflowStatusContent(state, trigger.stageId),
|
||||||
layer = ContextLayer.L0,
|
layer = ContextLayer.L0,
|
||||||
role = EntryRole.SYSTEM,
|
role = EntryRole.SYSTEM,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ import com.correx.core.events.types.InferenceRequestId
|
|||||||
import com.correx.core.events.types.SessionId
|
import com.correx.core.events.types.SessionId
|
||||||
import com.correx.core.events.types.StageId
|
import com.correx.core.events.types.StageId
|
||||||
import com.correx.core.inference.Embedder
|
import com.correx.core.inference.Embedder
|
||||||
|
import com.correx.core.inference.FinishReason
|
||||||
import com.correx.core.inference.InferenceRequest
|
import com.correx.core.inference.InferenceRequest
|
||||||
import com.correx.core.inference.InferenceRouter
|
import com.correx.core.inference.InferenceRouter
|
||||||
import com.correx.core.inference.ModelCapability
|
import com.correx.core.inference.ModelCapability
|
||||||
@@ -153,7 +154,18 @@ class DefaultRouterFacade(
|
|||||||
responseFormat = ResponseFormat.Text,
|
responseFormat = ResponseFormat.Text,
|
||||||
)
|
)
|
||||||
val inferenceResponse = provider.infer(inferenceRequest)
|
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
|
// Emit ROUTER turn event
|
||||||
emitChatTurn(
|
emitChatTurn(
|
||||||
@@ -162,19 +174,20 @@ class DefaultRouterFacade(
|
|||||||
tokensUsed = inferenceResponse.tokensUsed,
|
tokensUsed = inferenceResponse.tokensUsed,
|
||||||
)
|
)
|
||||||
|
|
||||||
if (mode == ChatMode.STEERING) {
|
val steeringEmitted = mode == ChatMode.STEERING && rawContent.isNotBlank()
|
||||||
|
if (steeringEmitted) {
|
||||||
val validationError = validateSteering?.invoke(content)
|
val validationError = validateSteering?.invoke(content)
|
||||||
if (validationError == null) {
|
if (validationError == null) {
|
||||||
emitSteeringNote(sessionId, content, effectiveStageId)
|
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) {
|
override suspend fun narrate(sessionId: SessionId, trigger: NarrationTrigger) {
|
||||||
val state = routerRepository.getRouterState(sessionId)
|
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)
|
val contextPack = routerContextBuilder.buildNarrationContext(state, trigger, config.tokenBudget)
|
||||||
|
|
||||||
|
|||||||
@@ -11,4 +11,10 @@ data class NarrationTrigger(
|
|||||||
val kind: String,
|
val kind: String,
|
||||||
/** Human-readable instruction injected as the final context entry for this narration. */
|
/** Human-readable instruction injected as the final context entry for this narration. */
|
||||||
val instruction: String,
|
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,
|
||||||
)
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user