kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature(event-sourcing): baseline for the architecture review fixes.

Browse Source 
- fixed the tool overlay in tui.
- fixed tool calling - added schemas.
- getting all sessionIds via event store.
- instead of sending all events on the replay - there's a new way for replaying.
- session snapshot is now a thing getting sent for previously created sessions.
- added logging to important parts.
- revert workflowId from WorkflowCompletedEvent.
This commit is contained in:
kami
2026-05-24 18:50:00 +04:00
parent f827685ed0
commit fc7b879891
43 changed files with 525 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/main/kotlin/com/correx/apps/server/Main.kt
+3
View File
@@ -129,6 +129,8 @@ fun main() {
providerRegistry = infraRegistry.asServerRegistry(),
defaultOrchestrationConfig = defaultOrchestrationConfig,
routerFacade = routerFacade,
orchestrationRepository = repositories.orchestrationRepository,
approvalRepository = repositories.approvalRepository,
)
log.info("==============================")
@@ -190,6 +192,7 @@ private fun buildRepositories(
DefaultEventReplayer(eventStore, SessionProjector(DefaultSessionReducer())),
),
artifactRepository = InfrastructureModule.createArtifactRepository(eventStore),
approvalRepository = InfrastructureModule.createApprovalRepository(eventStore),
)
private fun buildToolConfig(
apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt
+4
View File
@@ -2,10 +2,12 @@ package com.correx.apps.server
import com.correx.apps.server.registry.ProviderRegistry
import com.correx.apps.server.registry.WorkflowRegistry
import com.correx.core.approvals.DefaultApprovalRepository
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.stores.EventStore
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
import com.correx.core.kernel.orchestration.OrchestrationConfig
import com.correx.core.kernel.orchestration.OrchestrationRepository
import com.correx.core.router.RouterFacade
import com.correx.core.sessions.DefaultSessionRepository
@@ -18,4 +20,6 @@ data class ServerModule(
val providerRegistry: ProviderRegistry,
val defaultOrchestrationConfig: OrchestrationConfig,
val routerFacade: RouterFacade,
val orchestrationRepository: OrchestrationRepository,
val approvalRepository: DefaultApprovalRepository,
)
apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt
+8 -14
View File
@@ -19,7 +19,6 @@ 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.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.types.ArtifactId
class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) {
@@ -36,7 +35,6 @@ private object NoopArtifactStore : ArtifactStore {
@Suppress("CyclomaticComplexMethod")
suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: ArtifactStore): ServerMessage? =
when (val p = event.payload) {
is WorkflowStartedEvent -> mapWorkflowStarted(p)
is WorkflowCompletedEvent -> ServerMessage.SessionCompleted(sessionId = p.sessionId)
is WorkflowFailedEvent -> ServerMessage.SessionFailed(sessionId = p.sessionId, reason = p.reason)
is TransitionExecutedEvent -> ServerMessage.StageStarted(
@@ -91,9 +89,6 @@ suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: Artifa
else -> null
}
private fun mapWorkflowStarted(p: WorkflowStartedEvent): ServerMessage =
ServerMessage.SessionStarted(sessionId = p.sessionId, workflowId = p.workflowId)
private fun mapOrchestrationPaused(p: OrchestrationPausedEvent): ServerMessage {
val reason = if (p.reason == "APPROVAL_PENDING") PauseReason.APPROVAL_PENDING else PauseReason.USER_REQUESTED
return ServerMessage.SessionPaused(sessionId = p.sessionId, reason = reason)
@@ -103,15 +98,14 @@ private suspend fun mapInferenceCompleted(
event: StoredEvent,
p: InferenceCompletedEvent,
artifactStore: ArtifactStore,
): ServerMessage {
val text = runCatching {
artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: ""
}.getOrElse { "" }
return ServerMessage.InferenceCompleted(
): ServerMessage = runCatching {
artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: ""
}.getOrElse { "" }.run {
ServerMessage.InferenceCompleted(
sessionId = p.sessionId,
stageId = p.stageId,
outputSummary = text,
responseText = text,
outputSummary = this,
responseText = this,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
}
@@ -122,6 +116,6 @@ private fun mapApprovalRequested(p: ApprovalRequestedEvent): ServerMessage =
requestId = p.requestId,
tier = p.tier,
riskSummary = RiskSummaryDto("unknown", emptyList(), ""),
toolName = null,
preview = null,
toolName = p.toolName,
preview = p.preview,
)
apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt
+31 -3
View File
@@ -1,19 +1,47 @@
package com.correx.apps.server.bridge
import com.correx.apps.server.protocol.ServerMessage
import com.correx.core.approvals.DefaultApprovalRepository
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.orchestration.OrchestrationStatus
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.SessionId
import com.correx.core.kernel.orchestration.OrchestrationRepository
class SessionEventBridge(
private val eventStore: EventStore,
private val artifactStore: ArtifactStore,
private val orchestrationRepository: OrchestrationRepository,
private val approvalRepository: DefaultApprovalRepository,
private val send: suspend (ServerMessage) -> Unit,
) {
suspend fun replaySnapshot() {
eventStore.allEvents().toList()
.mapNotNull { domainEventToServerMessage(it, artifactStore) }
.forEach { send(it) }
eventStore.allSessionIds().forEach { sessionId ->
val orchState = orchestrationRepository.getState(sessionId)
if (orchState.status == OrchestrationStatus.IDLE) return@forEach
val approvalState = if (orchState.pendingApproval) {
approvalRepository.getApprovalState(sessionId)
} else null
val pendingRequest = approvalState?.requests?.values
?.firstOrNull { req ->
approvalState.decisions.values.none { it.requestId == req.id }
}
send(ServerMessage.SessionSnapshot(
sessionId = sessionId,
workflowId = orchState.workflowId,
status = orchState.status.name,
currentStageId = orchState.currentStageId?.value,
pauseReason = orchState.pauseReason,
pendingApproval = orchState.pendingApproval,
approvalRequestId = pendingRequest?.id?.value,
approvalTier = pendingRequest?.tier?.name,
approvalToolName = pendingRequest?.toolName,
approvalPreview = pendingRequest?.preview,
))
}
}
suspend fun streamLive(sessionId: SessionId) {
apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt
+1
View File
@@ -60,4 +60,5 @@ class LoggingEventStore(private val delegate: EventStore) : EventStore {
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = delegate.subscribe(sessionId)
override fun allEvents(): Sequence<StoredEvent> = delegate.allEvents()
override fun allSessionIds(): Set<SessionId> = delegate.allSessionIds().also { log.debug("got session ids from store: {}", it) }
}
apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt
+16
View File
@@ -4,6 +4,7 @@ import com.correx.core.approvals.Tier
import com.correx.core.events.types.ApprovalRequestId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import kotlinx.serialization.SerialName
import kotlinx.serialization.Serializable
@Serializable
@@ -20,6 +21,21 @@ sealed class ServerMessage {
@Serializable
data class SessionFailed(val sessionId: SessionId, val reason: String) : ServerMessage()
@Serializable
@SerialName("session_snapshot")
data class SessionSnapshot(
val sessionId: SessionId,
val workflowId: String,
val status: String,
val currentStageId: String?,
val pauseReason: String?,
val pendingApproval: Boolean,
val approvalRequestId: String?,
val approvalTier: String?,
val approvalToolName: String?,
val approvalPreview: String?,
) : ServerMessage()
@Serializable
data class StageStarted(val sessionId: SessionId, val stageId: StageId, val occurredAt: Long) : ServerMessage()
apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt
+16 -8
View File
@@ -38,6 +38,8 @@ class GlobalStreamHandler(private val module: ServerModule) {
eventStore = module.eventStore,
artifactStore = module.artifactStore,
send = sendFrame,
orchestrationRepository = module.orchestrationRepository,
approvalRepository = module.approvalRepository,
)
sendInitialSnapshot(session)
bridge.replaySnapshot()
@@ -104,6 +106,7 @@ class GlobalStreamHandler(private val module: ServerModule) {
is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported"))
is ClientMessage.ApprovalResponse ->
session.send(encodeError("ApprovalResponse must be sent to /sessions/{id}/stream"))
is ClientMessage.ChatInput -> Unit // handler deferred to subsequent task
}
}
@@ -150,14 +153,19 @@ class GlobalStreamHandler(private val module: ServerModule) {
)
}
}
liveStreamJobs.put(sessionId, session.launch {
val bridge = SessionEventBridge(
eventStore = module.eventStore,
artifactStore = module.artifactStore,
send = sendFrame,
)
bridge.streamLive(sessionId)
})?.cancel()
liveStreamJobs.put(
sessionId,
session.launch {
val bridge = SessionEventBridge(
eventStore = module.eventStore,
artifactStore = module.artifactStore,
send = sendFrame,
orchestrationRepository = module.orchestrationRepository,
approvalRepository = module.approvalRepository,
)
bridge.streamLive(sessionId)
},
)?.cancel()
val started = ServerMessage.SessionStarted(sessionId = sessionId, workflowId = msg.workflowId)
session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started)))
}
apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt
+1 -2
View File
@@ -79,11 +79,10 @@ class SessionStreamHandler(private val module: ServerModule) {
val domainDecision = msg.toDomainDecision(msg.requestId.value)
runCatching { module.orchestrator.submitApprovalDecision(msg.requestId, domainDecision) }
.onFailure {
log.error(
log.warn(
"submitApprovalDecision failed for request={}: {}",
msg.requestId.value,
it.message,
it,
)
}
}
apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt
-1
View File
@@ -83,7 +83,6 @@ class DomainEventMapperTest {
sessionId = sessionId,
terminalStageId = stageId,
totalStages = 1,
workflowId = workflowId,
),
)
val result = domainEventToServerMessage(event, noopStore)
apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt
+50 -8
View File
@@ -1,6 +1,9 @@
package com.correx.apps.server.bridge
import com.correx.apps.server.protocol.ServerMessage
import com.correx.core.approvals.ApprovalProjector
import com.correx.core.approvals.DefaultApprovalReducer
import com.correx.core.approvals.DefaultApprovalRepository
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.EventMetadata
import com.correx.core.events.events.EventPayload
@@ -14,6 +17,10 @@ import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer
import com.correx.core.kernel.orchestration.OrchestrationProjector
import com.correx.core.kernel.orchestration.OrchestrationRepository
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
@@ -61,17 +68,37 @@ class SessionEventBridgeTest {
override fun lastSequence(sessionId: SessionId): Long? = null
override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = liveFlow
override fun allEvents(): Sequence<StoredEvent> = allEventsList.asSequence()
override fun allSessionIds(): Set<SessionId> = allEventsList.map { it.metadata.sessionId }.toSet()
}
private val orchestrationRepository = OrchestrationRepository(
DefaultEventReplayer(
fakeEventStore(),
OrchestrationProjector(DefaultOrchestrationReducer()),
),
)
private val approvalRepository = DefaultApprovalRepository(
DefaultEventReplayer(
fakeEventStore(),
ApprovalProjector(DefaultApprovalReducer()),
),
)
@Test
fun `replaySnapshot sends mapped messages for all events`() = runTest {
val events = listOf(
storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
)
val store = fakeEventStore(allEventsList = events)
val sent = mutableListOf<ServerMessage>()
val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) }
val bridge = SessionEventBridge(
store,
noopArtifactStore,
orchestrationRepository,
approvalRepository,
) { sent.add(it) }
bridge.replaySnapshot()
@@ -84,11 +111,16 @@ class SessionEventBridgeTest {
fun `replaySnapshot skips unmapped events`() = runTest {
val events = listOf(
storedEvent(OrchestrationResumedEvent(sessionId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
)
val store = fakeEventStore(allEventsList = events)
val sent = mutableListOf<ServerMessage>()
val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) }
val bridge = SessionEventBridge(
store,
noopArtifactStore,
orchestrationRepository,
approvalRepository,
) { sent.add(it) }
bridge.replaySnapshot()
@@ -100,12 +132,17 @@ class SessionEventBridgeTest {
fun `streamLive sends mapped messages from live flow`() = runTest {
val events = listOf(
storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
)
val liveFlow: Flow<StoredEvent> = flow { events.forEach { emit(it) } }
val store = fakeEventStore(liveFlow = liveFlow)
val sent = mutableListOf<ServerMessage>()
val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) }
val bridge = SessionEventBridge(
store,
noopArtifactStore,
orchestrationRepository,
approvalRepository,
) { sent.add(it) }
bridge.streamLive(sessionId)
@@ -122,12 +159,17 @@ class SessionEventBridgeTest {
}
val store = fakeEventStore(liveFlow = liveFlow)
val sent = mutableListOf<ServerMessage>()
val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) }
val bridge = SessionEventBridge(
store,
noopArtifactStore,
orchestrationRepository,
approvalRepository,
) { sent.add(it) }
channel.send(storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L))
val job = launch { bridge.streamLive(sessionId) }
channel.send(storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L))
channel.send(storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L))
channel.close()
job.join()