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 ede731d9..a6d2dfae 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 @@ -1,55 +1,12 @@ package com.correx.apps.server.bridge -import com.correx.apps.server.protocol.AssessedIssueDto -import com.correx.apps.server.protocol.PauseReason -import com.correx.apps.server.protocol.ReviewFindingDto -import com.correx.apps.server.protocol.RiskSummaryDto import com.correx.apps.server.protocol.ServerMessage -import com.correx.apps.server.protocol.toDto import com.correx.core.artifactstore.ArtifactStore -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.ClarificationRequestedEvent -import com.correx.core.events.events.InferenceCompletedEvent -import com.correx.core.events.events.SessionNamedEvent -import com.correx.core.events.events.SessionWorkspaceBoundEvent -import com.correx.core.events.events.WorkflowStartedEvent -import com.correx.core.events.events.InferenceFailedEvent -import com.correx.core.events.events.InferenceStartedEvent -import com.correx.core.events.events.InferenceTimeoutEvent -import com.correx.core.events.events.ReviewFindingsRaisedEvent -import com.correx.core.events.events.RetryAttemptedEvent -import com.correx.core.events.events.ModelLoadedEvent -import com.correx.core.events.events.ModelUnloadedEvent -import com.correx.core.events.events.OrchestrationPausedEvent -import com.correx.core.events.events.OrchestrationResumedEvent -import com.correx.core.events.events.PreemptRedirectBlockedEvent -import com.correx.core.events.events.PreemptRedirectEvent -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.ToolExecutionCompletedEvent -import com.correx.core.events.events.ToolExecutionFailedEvent -import com.correx.core.events.events.ToolExecutionRejectedEvent -import com.correx.core.events.events.ToolCallAssessedEvent -import com.correx.core.events.events.ToolInvocationRequestedEvent -import com.correx.core.events.events.TransitionExecutedEvent -import com.correx.core.events.events.WorkflowCompletedEvent -import com.correx.core.events.events.TalkieNarrationEvent -import com.correx.core.events.events.WorkflowFailedEvent -import com.correx.core.events.events.WorkflowProposedEvent -import com.correx.core.events.risk.RiskAction import com.correx.core.events.types.ArtifactId import org.slf4j.LoggerFactory -private val log = LoggerFactory.getLogger("DomainEventMapper") +internal val log = LoggerFactory.getLogger("DomainEventMapper") class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) { suspend fun map(event: StoredEvent, sessionSequence: Long = 0L): ServerMessage? = @@ -62,379 +19,44 @@ private object NoopArtifactStore : ArtifactStore { override suspend fun flushBefore(commit: suspend () -> Unit) = commit() } -@Suppress("CyclomaticComplexMethod") +/** + * Outcome of one per-domain mapper. [Emit] means "this event is mine" — carrying either a + * [ServerMessage] or `null` (handled but deliberately not surfaced, e.g. transient bookkeeping + * events). [Skip] means "not my domain", so the dispatcher tries the next mapper. The distinction + * matters: chaining on a bare `null` would conflate suppression with non-ownership. + */ +internal sealed interface MapOutcome { + @JvmInline + value class Emit(val message: ServerMessage?) : MapOutcome + data object Skip : MapOutcome +} + +// Per-domain mappers, tried in order. Each owns a disjoint slice of the payload hierarchy and +// returns [MapOutcome.Skip] for anything outside it. Split across files by domain area to keep +// each mapper and its import list small — see StageInferenceEventMappers, ToolEventMappers, etc. +private val domainMappers: List MapOutcome> = listOf( + ::mapSessionEvent, + ::mapStageInferenceEvent, + ::mapToolEvent, + ::mapLifecycleEvent, +) + suspend fun domainEventToServerMessage( event: StoredEvent, artifactStore: ArtifactStore, sessionSequence: Long = 0L, ): ServerMessage? { - val seq = event.sequence - return when (val p = event.payload) { - is ChatSessionStartedEvent -> ServerMessage.SessionAnnounced( - sessionId = p.sessionId, - workflowId = "chat", - sequence = seq, - sessionSequence = sessionSequence, - ) - - is WorkflowStartedEvent -> ServerMessage.SessionAnnounced( - sessionId = p.sessionId, - workflowId = p.workflowId, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is SessionWorkspaceBoundEvent -> ServerMessage.SessionWorkspaceBound( - sessionId = p.sessionId, - workspaceRoot = p.workspaceRoot, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is SessionNamedEvent -> ServerMessage.SessionRenamed( - sessionId = p.sessionId, - name = p.name, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ChatTurnEvent -> ServerMessage.ChatTurn( - sessionId = p.sessionId, - turnId = p.turnId, - role = p.role.name, - content = p.content, - latencyMs = p.latencyMs, - totalTokens = p.tokensUsed?.totalTokens, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is WorkflowCompletedEvent -> ServerMessage.SessionCompleted( - sessionId = p.sessionId, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is WorkflowFailedEvent -> ServerMessage.SessionFailed( - sessionId = p.sessionId, - reason = p.reason, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is TransitionExecutedEvent -> ServerMessage.StageStarted( - sessionId = p.sessionId, - stageId = p.to, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - sequence = seq, - sessionSequence = sessionSequence, - ) - - is StageCompletedEvent -> ServerMessage.StageCompleted( - sessionId = p.sessionId, - stageId = p.stageId, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - sequence = seq, - sessionSequence = sessionSequence, - ) - - is StageFailedEvent -> ServerMessage.StageFailed( - sessionId = p.sessionId, - stageId = p.stageId, - reason = p.reason, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - sequence = seq, - sessionSequence = sessionSequence, - ) - - is OrchestrationPausedEvent -> mapOrchestrationPaused(p, seq, sessionSequence) - is OrchestrationResumedEvent -> ServerMessage.SessionResumed( - sessionId = p.sessionId, - stageId = p.stageId, - sequence = seq, - sessionSequence = sessionSequence, - ) - is InferenceStartedEvent -> ServerMessage.InferenceStarted( - sessionId = p.sessionId, - stageId = p.stageId, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is InferenceCompletedEvent -> mapInferenceCompleted(event, p, artifactStore, sessionSequence) - is InferenceTimeoutEvent -> ServerMessage.InferenceTimedOut( - sessionId = p.sessionId, - stageId = p.stageId, - elapsedMs = p.timeoutMs, - sequence = seq, - sessionSequence = sessionSequence, - ) - is InferenceFailedEvent -> ServerMessage.InferenceFailed( - sessionId = p.sessionId, - stageId = p.stageId, - reason = p.reason, - sequence = seq, - sessionSequence = sessionSequence, - ) - is RetryAttemptedEvent -> ServerMessage.RetryAttempted( - sessionId = p.sessionId, - stageId = p.stageId, - attemptNumber = p.attemptNumber, - maxAttempts = p.maxAttempts, - failureReason = p.failureReason, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ToolInvocationRequestedEvent -> ServerMessage.ToolStarted( - sessionId = p.sessionId, - toolName = p.toolName, - tier = p.tier, - params = prettyToolParams(p.request.parameters), - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ToolExecutionCompletedEvent -> ServerMessage.ToolCompleted( - sessionId = p.sessionId, - toolName = p.toolName, - outputSummary = p.receipt.outputSummary, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - diff = p.receipt.diff, - affectedEntities = p.receipt.affectedEntities, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ToolExecutionFailedEvent -> ServerMessage.ToolFailed( - sessionId = p.sessionId, - toolName = p.toolName, - reason = p.reason, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ToolExecutionRejectedEvent -> ServerMessage.ToolRejected( - sessionId = p.sessionId, - toolName = p.toolName, - reason = p.reason, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ToolCallAssessedEvent -> ServerMessage.ToolAssessed( - sessionId = p.sessionId, - stageId = p.stageId, - toolName = p.toolName, - disposition = p.disposition.name, - issues = p.issues.map { AssessedIssueDto(it.code, it.message, it.severity) }, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ReviewFindingsRaisedEvent -> ServerMessage.ReviewFindings( - sessionId = p.sessionId, - stageId = p.stageId, - verdict = p.verdict.name, - findings = p.findings.map { - ReviewFindingDto( - severity = it.severity.name, - confidence = it.confidence, - category = it.category, - target = it.target, - message = it.message, - suggestedFix = it.suggestedFix, - correctness = it.correctness, - ) - }, - blocked = p.blocked, - sequence = seq, - sessionSequence = sessionSequence, - ) - - is ApprovalRequestedEvent -> mapApprovalRequested(p, seq, sessionSequence) - is ClarificationRequestedEvent -> ServerMessage.ClarificationRequired( - sessionId = p.sessionId, - requestId = p.requestId, - stageId = p.stageId, - questions = p.questions, - sequence = seq, - sessionSequence = sessionSequence, - ) - is WorkflowProposedEvent -> ServerMessage.WorkflowProposed( - sessionId = p.sessionId, - proposalId = p.proposalId, - prompt = p.prompt, - candidates = p.candidates, - originalRequest = p.originalRequest, - sequence = seq, - sessionSequence = sessionSequence, - ) - is ApprovalDecisionResolvedEvent -> ServerMessage.ApprovalResolved( - // ApprovalDecisionResolvedEvent has no sessionId on its payload — read it from the event envelope - sessionId = event.metadata.sessionId, - requestId = p.requestId, - outcome = p.outcome.name, - reason = p.reason, - sequence = seq, - sessionSequence = sessionSequence, - ) - is TalkieNarrationEvent -> ServerMessage.Narration( - sessionId = p.sessionId, - content = p.content, - stageId = p.stageId, - latencyMs = p.latencyMs, - totalTokens = p.tokensUsed?.totalTokens, - sequence = seq, - sessionSequence = sessionSequence, - ) - is ArtifactCreatedEvent -> ServerMessage.ArtifactCreated( - sessionId = p.sessionId, - stageId = p.stageId, - artifactId = p.artifactId, - 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, - loaded = true, - ) - is ModelUnloadedEvent -> ServerMessage.ModelChanged( - modelId = p.modelId, - providerId = p.providerId.value, - loaded = false, - ) - // Internal slot→CAS-hash bookkeeping (F-007 durable bridge); no operator-facing surface. - is ArtifactContentStoredEvent -> null - - // Freestyle graph-rerouting bookkeeping. The deterministic record is in place; a dedicated - // operator surface ships with the LLM-proposal + approval-confirm front-half. - is PreemptRedirectEvent -> null - is PreemptRedirectBlockedEvent -> null - - else -> { - log.debug( - "DomainEventMapper: unmapped payload type={} sessionId={} sequence={}", - p::class.simpleName, - event.metadata.sessionId, - event.sequence, - ) - null + for (mapper in domainMappers) { + when (val outcome = mapper(event, artifactStore, sessionSequence)) { + is MapOutcome.Emit -> return outcome.message + MapOutcome.Skip -> Unit } } -} - -private fun mapOrchestrationPaused( - p: OrchestrationPausedEvent, - seq: Long, - sessionSequence: Long, -): ServerMessage { - val reason = when (p.reason) { - "APPROVAL_PENDING" -> PauseReason.APPROVAL_PENDING - "CLARIFICATION_PENDING" -> PauseReason.CLARIFICATION_PENDING - "ABANDONED_STALE" -> PauseReason.ABANDONED_STALE - else -> PauseReason.USER_REQUESTED - } - return ServerMessage.SessionPaused( - sessionId = p.sessionId, - reason = reason, - sequence = seq, - sessionSequence = sessionSequence, + log.debug( + "DomainEventMapper: unmapped payload type={} sessionId={} sequence={}", + event.payload::class.simpleName, + event.metadata.sessionId, + event.sequence, ) + return null } - -private suspend fun mapInferenceCompleted( - event: StoredEvent, - p: InferenceCompletedEvent, - artifactStore: ArtifactStore, - sessionSequence: Long, -): ServerMessage { - val response = runCatching { - artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: "" - }.getOrElse { "" } - val reasoning = p.reasoningArtifactId?.let { id -> - runCatching { artifactStore.get(id)?.toString(Charsets.UTF_8) ?: "" }.getOrElse { "" } - } ?: "" - return ServerMessage.InferenceCompleted( - sessionId = p.sessionId, - stageId = p.stageId, - outputSummary = response, - responseText = response, - reasoning = reasoning, - occurredAt = event.metadata.timestamp.toEpochMilliseconds(), - totalTokens = p.tokensUsed.totalTokens, - sequence = event.sequence, - sessionSequence = sessionSequence, - ) -} - -private fun mapApprovalRequested( - p: ApprovalRequestedEvent, - seq: Long, - sessionSequence: Long, -): ServerMessage = - ServerMessage.ApprovalRequired( - sessionId = p.sessionId, - requestId = p.requestId, - tier = p.tier, - riskSummary = p.riskSummary?.toDto() ?: RiskSummaryDto( - level = p.tier.name, - factors = emptyList(), - recommendedAction = RiskAction.PROMPT_USER.name, - rationale = emptyList(), - ), - toolName = p.toolName, - preview = p.preview, - sequence = seq, - sessionSequence = sessionSequence, - ) - -/** - * Render a tool call's raw parameter map into compact, human-readable "key=value" cells for the - * client's tool-call row. Primary keys (path/command/query/…) lead so the most salient argument is - * first; long values are truncated and multi-line values collapsed so the WS frame stays small. - */ -internal fun prettyToolParams(parameters: Map): List { - if (parameters.isEmpty()) return emptyList() - val primary = listOf("command", "path", "query", "url", "pattern", "content", "operation") - val ordered = parameters.entries.sortedWith( - compareBy({ primary.indexOf(it.key).let { i -> if (i < 0) primary.size else i } }, { it.key }), - ) - return ordered.take(MAX_PARAM_CELLS).map { (k, v) -> "$k=${formatParamValue(v)}" } -} - -private fun formatParamValue(value: Any?): String { - val raw = when (value) { - null -> "null" - is String -> value - is Collection<*> -> value.joinToString(", ", prefix = "[", postfix = "]") { formatParamValue(it) } - else -> value.toString() - } - val flattened = raw.replace('\n', ' ').replace('\r', ' ').trim() - val clipped = if (flattened.length > MAX_PARAM_VALUE_LEN) flattened.take(MAX_PARAM_VALUE_LEN) + "…" else flattened - // Quote strings that carry whitespace so the boundary of the value is unambiguous in the row. - return if (value is String && clipped.any { it.isWhitespace() }) "\"$clipped\"" else clipped -} - -private const val MAX_PARAM_CELLS = 5 -private const val MAX_PARAM_VALUE_LEN = 80 diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/LifecycleEventMappers.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/LifecycleEventMappers.kt new file mode 100644 index 00000000..91fe9202 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/LifecycleEventMappers.kt @@ -0,0 +1,150 @@ +package com.correx.apps.server.bridge + +import com.correx.apps.server.protocol.RiskSummaryDto +import com.correx.apps.server.protocol.ServerMessage +import com.correx.apps.server.protocol.toDto +import com.correx.core.artifactstore.ArtifactStore +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.ClarificationRequestedEvent +import com.correx.core.events.events.ExecutionPlanLockedEvent +import com.correx.core.events.events.ModelLoadedEvent +import com.correx.core.events.events.ModelUnloadedEvent +import com.correx.core.events.events.PreemptRedirectBlockedEvent +import com.correx.core.events.events.PreemptRedirectEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.TalkieNarrationEvent +import com.correx.core.events.events.WorkflowProposedEvent +import com.correx.core.events.risk.RiskAction + +/** + * Approval / clarification / workflow-proposal / narration / artifact / model / preempt events — + * the remaining operator-facing (or deliberately suppressed) lifecycle surface. Branches that + * return `null` are handled-but-not-surfaced (transient bookkeeping); wrapping them in + * [MapOutcome.Emit] keeps them from falling through to the dispatcher's "unmapped" log. + */ +@Suppress("UnusedParameter", "LongMethod", "CyclomaticComplexMethod") +internal suspend fun mapLifecycleEvent( + event: StoredEvent, + artifactStore: ArtifactStore, + sessionSequence: Long, +): MapOutcome { + val seq = event.sequence + val msg: ServerMessage? = when (val p = event.payload) { + is ApprovalRequestedEvent -> mapApprovalRequested(p, seq, sessionSequence) + is ClarificationRequestedEvent -> ServerMessage.ClarificationRequired( + sessionId = p.sessionId, + requestId = p.requestId, + stageId = p.stageId, + questions = p.questions, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is WorkflowProposedEvent -> ServerMessage.WorkflowProposed( + sessionId = p.sessionId, + proposalId = p.proposalId, + prompt = p.prompt, + candidates = p.candidates, + originalRequest = p.originalRequest, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ApprovalDecisionResolvedEvent -> ServerMessage.ApprovalResolved( + // ApprovalDecisionResolvedEvent has no sessionId on its payload — read it from the event envelope + sessionId = event.metadata.sessionId, + requestId = p.requestId, + outcome = p.outcome.name, + reason = p.reason, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is TalkieNarrationEvent -> ServerMessage.Narration( + sessionId = p.sessionId, + content = p.content, + stageId = p.stageId, + latencyMs = p.latencyMs, + totalTokens = p.tokensUsed?.totalTokens, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ArtifactCreatedEvent -> ServerMessage.ArtifactCreated( + sessionId = p.sessionId, + stageId = p.stageId, + artifactId = p.artifactId, + 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, + loaded = true, + ) + + is ModelUnloadedEvent -> ServerMessage.ModelChanged( + modelId = p.modelId, + providerId = p.providerId.value, + loaded = false, + ) + + // Internal slot→CAS-hash bookkeeping (F-007 durable bridge); no operator-facing surface. + is ArtifactContentStoredEvent -> null + + // Freestyle graph-rerouting bookkeeping. The deterministic record is in place; a dedicated + // operator surface ships with the LLM-proposal + approval-confirm front-half. + is PreemptRedirectEvent -> null + is PreemptRedirectBlockedEvent -> null + + else -> return MapOutcome.Skip + } + return MapOutcome.Emit(msg) +} + +private fun mapApprovalRequested( + p: ApprovalRequestedEvent, + seq: Long, + sessionSequence: Long, +): ServerMessage = + ServerMessage.ApprovalRequired( + sessionId = p.sessionId, + requestId = p.requestId, + tier = p.tier, + riskSummary = p.riskSummary?.toDto() ?: RiskSummaryDto( + level = p.tier.name, + factors = emptyList(), + recommendedAction = RiskAction.PROMPT_USER.name, + rationale = emptyList(), + ), + toolName = p.toolName, + preview = p.preview, + sequence = seq, + sessionSequence = sessionSequence, + ) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventMappers.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventMappers.kt new file mode 100644 index 00000000..b53f1984 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventMappers.kt @@ -0,0 +1,78 @@ +package com.correx.apps.server.bridge + +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.ChatSessionStartedEvent +import com.correx.core.events.events.ChatTurnEvent +import com.correx.core.events.events.SessionNamedEvent +import com.correx.core.events.events.SessionWorkspaceBoundEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.events.WorkflowStartedEvent + +/** Session lifecycle + chat-turn events. [artifactStore] is unused here but kept for a uniform mapper signature. */ +@Suppress("UnusedParameter", "LongMethod") +internal suspend fun mapSessionEvent( + event: StoredEvent, + artifactStore: ArtifactStore, + sessionSequence: Long, +): MapOutcome { + val seq = event.sequence + val msg = when (val p = event.payload) { + is ChatSessionStartedEvent -> ServerMessage.SessionAnnounced( + sessionId = p.sessionId, + workflowId = "chat", + sequence = seq, + sessionSequence = sessionSequence, + ) + + is WorkflowStartedEvent -> ServerMessage.SessionAnnounced( + sessionId = p.sessionId, + workflowId = p.workflowId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is SessionWorkspaceBoundEvent -> ServerMessage.SessionWorkspaceBound( + sessionId = p.sessionId, + workspaceRoot = p.workspaceRoot, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is SessionNamedEvent -> ServerMessage.SessionRenamed( + sessionId = p.sessionId, + name = p.name, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ChatTurnEvent -> ServerMessage.ChatTurn( + sessionId = p.sessionId, + turnId = p.turnId, + role = p.role.name, + content = p.content, + latencyMs = p.latencyMs, + totalTokens = p.tokensUsed?.totalTokens, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is WorkflowCompletedEvent -> ServerMessage.SessionCompleted( + sessionId = p.sessionId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is WorkflowFailedEvent -> ServerMessage.SessionFailed( + sessionId = p.sessionId, + reason = p.reason, + sequence = seq, + sessionSequence = sessionSequence, + ) + + else -> return MapOutcome.Skip + } + return MapOutcome.Emit(msg) +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/StageInferenceEventMappers.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/StageInferenceEventMappers.kt new file mode 100644 index 00000000..ee0492ac --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/StageInferenceEventMappers.kt @@ -0,0 +1,141 @@ +package com.correx.apps.server.bridge + +import com.correx.apps.server.protocol.PauseReason +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.InferenceFailedEvent +import com.correx.core.events.events.InferenceStartedEvent +import com.correx.core.events.events.InferenceTimeoutEvent +import com.correx.core.events.events.OrchestrationPausedEvent +import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.RetryAttemptedEvent +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 + +/** Stage transition + inference lifecycle events (some carry timestamps / need the artifact store). */ +@Suppress("LongMethod") +internal suspend fun mapStageInferenceEvent( + event: StoredEvent, + artifactStore: ArtifactStore, + sessionSequence: Long, +): MapOutcome { + val seq = event.sequence + val msg = when (val p = event.payload) { + is TransitionExecutedEvent -> ServerMessage.StageStarted( + sessionId = p.sessionId, + stageId = p.to, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, + ) + + is StageCompletedEvent -> ServerMessage.StageCompleted( + sessionId = p.sessionId, + stageId = p.stageId, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, + ) + + is StageFailedEvent -> ServerMessage.StageFailed( + sessionId = p.sessionId, + stageId = p.stageId, + reason = p.reason, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, + ) + + is OrchestrationPausedEvent -> mapOrchestrationPaused(p, seq, sessionSequence) + is OrchestrationResumedEvent -> ServerMessage.SessionResumed( + sessionId = p.sessionId, + stageId = p.stageId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is InferenceStartedEvent -> ServerMessage.InferenceStarted( + sessionId = p.sessionId, + stageId = p.stageId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is InferenceCompletedEvent -> mapInferenceCompleted(event, p, artifactStore, sessionSequence) + is InferenceTimeoutEvent -> ServerMessage.InferenceTimedOut( + sessionId = p.sessionId, + stageId = p.stageId, + elapsedMs = p.timeoutMs, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is InferenceFailedEvent -> ServerMessage.InferenceFailed( + sessionId = p.sessionId, + stageId = p.stageId, + reason = p.reason, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is RetryAttemptedEvent -> ServerMessage.RetryAttempted( + sessionId = p.sessionId, + stageId = p.stageId, + attemptNumber = p.attemptNumber, + maxAttempts = p.maxAttempts, + failureReason = p.failureReason, + sequence = seq, + sessionSequence = sessionSequence, + ) + + else -> return MapOutcome.Skip + } + return MapOutcome.Emit(msg) +} + +private fun mapOrchestrationPaused( + p: OrchestrationPausedEvent, + seq: Long, + sessionSequence: Long, +): ServerMessage { + val reason = when (p.reason) { + "APPROVAL_PENDING" -> PauseReason.APPROVAL_PENDING + "CLARIFICATION_PENDING" -> PauseReason.CLARIFICATION_PENDING + "ABANDONED_STALE" -> PauseReason.ABANDONED_STALE + else -> PauseReason.USER_REQUESTED + } + return ServerMessage.SessionPaused( + sessionId = p.sessionId, + reason = reason, + sequence = seq, + sessionSequence = sessionSequence, + ) +} + +private suspend fun mapInferenceCompleted( + event: StoredEvent, + p: InferenceCompletedEvent, + artifactStore: ArtifactStore, + sessionSequence: Long, +): ServerMessage { + val response = runCatching { + artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: "" + }.getOrElse { "" } + val reasoning = p.reasoningArtifactId?.let { id -> + runCatching { artifactStore.get(id)?.toString(Charsets.UTF_8) ?: "" }.getOrElse { "" } + } ?: "" + return ServerMessage.InferenceCompleted( + sessionId = p.sessionId, + stageId = p.stageId, + outputSummary = response, + responseText = response, + reasoning = reasoning, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + totalTokens = p.tokensUsed.totalTokens, + sequence = event.sequence, + sessionSequence = sessionSequence, + ) +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/ToolEventMappers.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/ToolEventMappers.kt new file mode 100644 index 00000000..ca94c6e8 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/ToolEventMappers.kt @@ -0,0 +1,125 @@ +package com.correx.apps.server.bridge + +import com.correx.apps.server.protocol.AssessedIssueDto +import com.correx.apps.server.protocol.ReviewFindingDto +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.ReviewFindingsRaisedEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.ToolCallAssessedEvent +import com.correx.core.events.events.ToolExecutionCompletedEvent +import com.correx.core.events.events.ToolExecutionFailedEvent +import com.correx.core.events.events.ToolExecutionRejectedEvent +import com.correx.core.events.events.ToolInvocationRequestedEvent + +/** Tool invocation / execution / assessment + review-findings events. */ +@Suppress("UnusedParameter", "LongMethod") +internal suspend fun mapToolEvent( + event: StoredEvent, + artifactStore: ArtifactStore, + sessionSequence: Long, +): MapOutcome { + val seq = event.sequence + val msg = when (val p = event.payload) { + is ToolInvocationRequestedEvent -> ServerMessage.ToolStarted( + sessionId = p.sessionId, + toolName = p.toolName, + tier = p.tier, + params = prettyToolParams(p.request.parameters), + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ToolExecutionCompletedEvent -> ServerMessage.ToolCompleted( + sessionId = p.sessionId, + toolName = p.toolName, + outputSummary = p.receipt.outputSummary, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + diff = p.receipt.diff, + affectedEntities = p.receipt.affectedEntities, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ToolExecutionFailedEvent -> ServerMessage.ToolFailed( + sessionId = p.sessionId, + toolName = p.toolName, + reason = p.reason, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ToolExecutionRejectedEvent -> ServerMessage.ToolRejected( + sessionId = p.sessionId, + toolName = p.toolName, + reason = p.reason, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ToolCallAssessedEvent -> ServerMessage.ToolAssessed( + sessionId = p.sessionId, + stageId = p.stageId, + toolName = p.toolName, + disposition = p.disposition.name, + issues = p.issues.map { AssessedIssueDto(it.code, it.message, it.severity) }, + occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, + ) + + is ReviewFindingsRaisedEvent -> ServerMessage.ReviewFindings( + sessionId = p.sessionId, + stageId = p.stageId, + verdict = p.verdict.name, + findings = p.findings.map { + ReviewFindingDto( + severity = it.severity.name, + confidence = it.confidence, + category = it.category, + target = it.target, + message = it.message, + suggestedFix = it.suggestedFix, + correctness = it.correctness, + ) + }, + blocked = p.blocked, + sequence = seq, + sessionSequence = sessionSequence, + ) + + else -> return MapOutcome.Skip + } + return MapOutcome.Emit(msg) +} + +/** + * Render a tool call's raw parameter map into compact, human-readable "key=value" cells for the + * client's tool-call row. Primary keys (path/command/query/…) lead so the most salient argument is + * first; long values are truncated and multi-line values collapsed so the WS frame stays small. + */ +internal fun prettyToolParams(parameters: Map): List { + if (parameters.isEmpty()) return emptyList() + val primary = listOf("command", "path", "query", "url", "pattern", "content", "operation") + val ordered = parameters.entries.sortedWith( + compareBy({ primary.indexOf(it.key).let { i -> if (i < 0) primary.size else i } }, { it.key }), + ) + return ordered.take(MAX_PARAM_CELLS).map { (k, v) -> "$k=${formatParamValue(v)}" } +} + +private fun formatParamValue(value: Any?): String { + val raw = when (value) { + null -> "null" + is String -> value + is Collection<*> -> value.joinToString(", ", prefix = "[", postfix = "]") { formatParamValue(it) } + else -> value.toString() + } + val flattened = raw.replace('\n', ' ').replace('\r', ' ').trim() + val clipped = if (flattened.length > MAX_PARAM_VALUE_LEN) flattened.take(MAX_PARAM_VALUE_LEN) + "…" else flattened + // Quote strings that carry whitespace so the boundary of the value is unambiguous in the row. + return if (value is String && clipped.any { it.isWhitespace() }) "\"$clipped\"" else clipped +} + +private const val MAX_PARAM_CELLS = 5 +private const val MAX_PARAM_VALUE_LEN = 80