refactor(server): decompose DomainEventMapper god-when into per-domain files
DomainEventMapper.kt had grown into a single ~280-line `when` over every
domain event type with a ~40-line flat import list ('god mapper' smell).
Split the when body by domain area into four files, each a suspend fun
returning a MapOutcome sentinel:
- SessionEventMappers.kt — session/workflow lifecycle + chat turns
- StageInferenceEventMappers.kt — stage transitions + inference lifecycle
- ToolEventMappers.kt — tool exec/assessment + review findings (+ prettyToolParams)
- LifecycleEventMappers.kt — approval/clarification/proposal/narration/artifact/model/preempt
The dispatcher chains them; MapOutcome.Emit(msg?) vs MapOutcome.Skip
preserves the null-vs-unhandled distinction (a suppressed event like
ArtifactValidating returns Emit(null), not Skip). Public entry points
(DomainEventMapper class, domainEventToServerMessage, prettyToolParams)
unchanged — no behavior change, pure maintainability. Each per-domain
import list is now local to its file. DomainEventMapperTest green.
Vikunja #31.
This commit is contained in:
@@ -1,55 +1,12 @@
|
|||||||
package com.correx.apps.server.bridge
|
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.ServerMessage
|
||||||
import com.correx.apps.server.protocol.toDto
|
|
||||||
import com.correx.core.artifactstore.ArtifactStore
|
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.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 com.correx.core.events.types.ArtifactId
|
||||||
import org.slf4j.LoggerFactory
|
import org.slf4j.LoggerFactory
|
||||||
|
|
||||||
private val log = LoggerFactory.getLogger("DomainEventMapper")
|
internal val log = LoggerFactory.getLogger("DomainEventMapper")
|
||||||
|
|
||||||
class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) {
|
class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) {
|
||||||
suspend fun map(event: StoredEvent, sessionSequence: Long = 0L): ServerMessage? =
|
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()
|
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<suspend (StoredEvent, ArtifactStore, Long) -> MapOutcome> = listOf(
|
||||||
|
::mapSessionEvent,
|
||||||
|
::mapStageInferenceEvent,
|
||||||
|
::mapToolEvent,
|
||||||
|
::mapLifecycleEvent,
|
||||||
|
)
|
||||||
|
|
||||||
suspend fun domainEventToServerMessage(
|
suspend fun domainEventToServerMessage(
|
||||||
event: StoredEvent,
|
event: StoredEvent,
|
||||||
artifactStore: ArtifactStore,
|
artifactStore: ArtifactStore,
|
||||||
sessionSequence: Long = 0L,
|
sessionSequence: Long = 0L,
|
||||||
): ServerMessage? {
|
): ServerMessage? {
|
||||||
val seq = event.sequence
|
for (mapper in domainMappers) {
|
||||||
return when (val p = event.payload) {
|
when (val outcome = mapper(event, artifactStore, sessionSequence)) {
|
||||||
is ChatSessionStartedEvent -> ServerMessage.SessionAnnounced(
|
is MapOutcome.Emit -> return outcome.message
|
||||||
sessionId = p.sessionId,
|
MapOutcome.Skip -> Unit
|
||||||
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(
|
log.debug(
|
||||||
"DomainEventMapper: unmapped payload type={} sessionId={} sequence={}",
|
"DomainEventMapper: unmapped payload type={} sessionId={} sequence={}",
|
||||||
p::class.simpleName,
|
event.payload::class.simpleName,
|
||||||
event.metadata.sessionId,
|
event.metadata.sessionId,
|
||||||
event.sequence,
|
event.sequence,
|
||||||
)
|
)
|
||||||
null
|
return null
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
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,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
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<String, Any>): List<String> {
|
|
||||||
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
|
|
||||||
|
|||||||
@@ -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,
|
||||||
|
)
|
||||||
@@ -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)
|
||||||
|
}
|
||||||
+141
@@ -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,
|
||||||
|
)
|
||||||
|
}
|
||||||
@@ -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<String, Any>): List<String> {
|
||||||
|
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
|
||||||
Reference in New Issue
Block a user