diff --git a/.gitignore b/.gitignore index 0589c94e..19bbf450 100644 --- a/.gitignore +++ b/.gitignore @@ -68,4 +68,5 @@ bin/ /code-analysis/ /.infra/sonarqube/docker-compose.yml /logs -log4j2.properties \ No newline at end of file +log4j2.properties +/docs/specs/ diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index f66d97bf..76ae7921 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -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( diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index b1dac07a..e6447520 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -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, ) 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 aa2418fb..c5a4715f 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 @@ -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, ) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt index 246c9295..69d374dc 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt @@ -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) { diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt b/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt index c651dcd8..93bf7533 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt @@ -60,4 +60,5 @@ class LoggingEventStore(private val delegate: EventStore) : EventStore { override fun subscribe(sessionId: SessionId): Flow = delegate.subscribe(sessionId) override fun allEvents(): Sequence = delegate.allEvents() + override fun allSessionIds(): Set = delegate.allSessionIds().also { log.debug("got session ids from store: {}", it) } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index 6006ff08..672d76be 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -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() diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt index 1fe5e5ae..fa20c2b0 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -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))) } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt index ec3c6cda..1f51767a 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt @@ -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, ) } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt index 9f746e16..c9571520 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt @@ -83,7 +83,6 @@ class DomainEventMapperTest { sessionId = sessionId, terminalStageId = stageId, totalStages = 1, - workflowId = workflowId, ), ) val result = domainEventToServerMessage(event, noopStore) diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt index 34f7c03c..c3f6b5c9 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt @@ -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 = liveFlow override fun allEvents(): Sequence = allEventsList.asSequence() + override fun allSessionIds(): Set = 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() - 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() - 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 = flow { events.forEach { emit(it) } } val store = fakeEventStore(liveFlow = liveFlow) val sent = mutableListOf() - 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() - 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() diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt index 194bab07..8d814872 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt @@ -25,6 +25,7 @@ import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking +import org.slf4j.LoggerFactory import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent private const val DEFAULT_PORT = 8080 @@ -32,6 +33,8 @@ private const val STATUS_HEIGHT = 1 private const val TOP_ROW_HEIGHT = 9 private const val INPUT_HEIGHT = 4 +private val log = LoggerFactory.getLogger("com.correx.apps.tui.main") + fun main(args: Array) { val host = args.getOrElse(0) { "localhost" } val port = args.getOrElse(1) { DEFAULT_PORT.toString() }.toIntOrNull() ?: DEFAULT_PORT @@ -43,6 +46,13 @@ fun main(args: Array) { TuiRunner.create().use { runner -> fun dispatch(action: Action) { val (next, effects) = RootReducer.reduce(state, action) + log.debug("got connection state after reducers: {}", next.connection) + log.debug("got approval state after reducers: {}", next.approval) + log.debug("got input state after reducers: {}", next.input) + log.debug( + "got session state after reducers: {}", + next.sessions.sessions.firstOrNull() { it.status != "COMPLETED" }, + ) state = next effects.forEach { effect -> effectScope.launch { diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt index 0e86fcb5..79231836 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/RouterPanel.kt @@ -27,10 +27,10 @@ fun routerPanelWidget(state: TuiState): Paragraph { state.approvalOverlayVisible && state.approval.active != null -> { val a = state.approval.active buildList { - add(Line.from(Span.styled("╭─ approval [ctrl+h to hide] ─────────────╮", Style.create().yellow()))) - val tierLine = "│ ⚠ T${a.tier} │ ${a.toolName ?: ""} │ ${a.riskSummary.take(12)}... │" + add(Line.from(Span.styled("╭─ approval [alt+h to hide] ─────────────╮", Style.create().yellow()))) + val tierLine = "│ ⚠ ${a.tier} │ ${a.toolName ?: "no tool name"} │ ${a.riskSummary.take(12)}... │" add(Line.from(Span.styled(tierLine, Style.create().yellow()))) - val previewLine = "│ ${(a.preview ?: "").take(40)}… │" + val previewLine = "│ ${(a.preview ?: "no preview").take(40)}… │" add(Line.from(Span.styled(previewLine, dimStyle))) add(Line.from(Span.styled("╰──────────────────────────────────────────╯", Style.create().yellow()))) } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/input/Action.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/input/Action.kt index b3f46767..90e1d8bd 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/input/Action.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/input/Action.kt @@ -24,4 +24,5 @@ sealed interface Action { data object ToggleEventOverlay : Action data object EnterSteer : Action data object CycleMode : Action + data object NoOp : Action } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/input/TambouiKeyMapper.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/input/TambouiKeyMapper.kt index e61f1a57..c6750d58 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/input/TambouiKeyMapper.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/input/TambouiKeyMapper.kt @@ -2,12 +2,24 @@ package com.correx.apps.tui.input import com.correx.apps.tui.KeyEvent import dev.tamboui.tui.event.KeyCode +import org.slf4j.LoggerFactory import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent +private val log = LoggerFactory.getLogger("com.correx.apps.tui.input") + @Suppress("CyclomaticComplexMethod", "ReturnCount") fun mapKey(event: TambouiKeyEvent): KeyEvent? { + log.debug("got event: {}", event) if (event.isCtrlC) return KeyEvent.Quit - if (event.hasAlt()) return null + if (event.hasAlt()) { + return when { + event.isChar('h') -> KeyEvent.ToggleApprovalOverlay.also { + log.debug("got event for toggle approval") + } + + else -> null + } + } if (event.hasCtrl()) { return when { event.isChar('q') -> KeyEvent.Quit @@ -16,7 +28,8 @@ fun mapKey(event: TambouiKeyEvent): KeyEvent? { event.isChar('a') -> KeyEvent.Approve event.isChar('r') -> KeyEvent.Reject event.isChar('s') -> KeyEvent.Steer - event.isChar('h') -> KeyEvent.ToggleApprovalOverlay + + event.isChar('e') -> KeyEvent.ToggleEventOverlay else -> null } @@ -32,6 +45,7 @@ fun mapKey(event: TambouiKeyEvent): KeyEvent? { val ch = event.character() if (ch.isISOControl()) null else KeyEvent.CharInput(ch) } + else -> null } } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/ApprovalReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/ApprovalReducer.kt index 30b99d81..dd936bd3 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/ApprovalReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/ApprovalReducer.kt @@ -8,6 +8,9 @@ import com.correx.apps.tui.state.ApprovalInfo import com.correx.apps.tui.state.ApprovalState import com.correx.apps.tui.state.InputMode import com.correx.core.events.types.ApprovalRequestId +import org.slf4j.LoggerFactory + +private val log = LoggerFactory.getLogger(ApprovalReducer::class.simpleName) object ApprovalReducer { fun reduce( @@ -31,6 +34,7 @@ object ApprovalReducer { approval to emptyList() } } + is Action.RejectActive -> { val active = approval.active if (active != null) { @@ -47,13 +51,19 @@ object ApprovalReducer { approval to emptyList() } } + is Action.SubmitInput -> if (inputMode == InputMode.STEER) { ApprovalState(active = null) to emptyList() } else { approval to emptyList() } + is Action.ServerEventReceived -> when (val msg = action.message) { is ServerMessage.ApprovalRequired -> { + log.debug( + "ApprovalRequired received requestId={} sessionId={} toolName={}", + msg.requestId.value, msg.sessionId.value, msg.toolName, + ) val info = ApprovalInfo( requestId = msg.requestId.value, sessionId = msg.sessionId.value, @@ -64,8 +74,27 @@ object ApprovalReducer { ) ApprovalState(active = info) to emptyList() } + + is ServerMessage.SessionSnapshot -> { + val requestId = msg.approvalRequestId + if (msg.pendingApproval && requestId != null) { + val info = ApprovalInfo( + requestId = requestId, + sessionId = msg.sessionId.value, + tier = msg.approvalTier ?: "T2", + riskSummary = "unknown", + toolName = msg.approvalToolName, + preview = msg.approvalPreview, + ) + ApprovalState(active = info) to emptyList() + } else { + approval to emptyList() + } + } + else -> approval to emptyList() } + else -> approval to emptyList() } } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt index fb5b2e67..381ccc8b 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt @@ -22,12 +22,14 @@ object InputReducer { }, inputBuffer = "", ) to emptyList() + is Action.OpenNewSessionPrompt -> state.copy(inputMode = InputMode.ROUTER, inputBuffer = "") to emptyList() is Action.OpenSteeringPrompt -> if (state.approval.active != null) { state.copy(inputMode = InputMode.STEER, inputBuffer = "") to emptyList() } else { state to emptyList() } + is Action.EnterSteer -> state.copy(inputMode = InputMode.STEER, inputBuffer = "") to emptyList() is Action.OpenFilter -> state.copy(inputMode = InputMode.FILTER, inputBuffer = "") to emptyList() is Action.AppendChar -> state.copy(inputBuffer = state.inputBuffer + action.ch) to emptyList() @@ -44,6 +46,7 @@ object InputReducer { state.copy(inputMode = InputMode.ROUTER, inputBuffer = "") to emptyList() } } + InputMode.STEER -> { val text = state.inputBuffer.trim() val active = state.approval.active @@ -61,9 +64,11 @@ object InputReducer { state.copy(inputMode = InputMode.ROUTER, inputBuffer = "") to emptyList() } } + InputMode.FILTER -> state.copy(inputMode = InputMode.ROUTER, inputBuffer = "") to emptyList() InputMode.NAVIGATE -> state to emptyList() } + else -> state to emptyList() } } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt index c257c74e..88c89f34 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt @@ -1,7 +1,11 @@ package com.correx.apps.tui.reducer +import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.tui.input.Action import com.correx.apps.tui.state.TuiState +import org.slf4j.LoggerFactory + +private val log = LoggerFactory.getLogger(RootReducer::class.java.name) object RootReducer { fun reduce( @@ -9,13 +13,25 @@ object RootReducer { action: Action, clock: () -> Long = System::currentTimeMillis, ): Pair> { + log.debug("action={}", action::class.simpleName) val quitEffects: List = if (action is Action.Quit) listOf(Effect.Quit) else emptyList() val prevInputMode = state.inputMode val prevInputBuffer = state.inputBuffer + val approvalAction = if ( + action is Action.ServerEventReceived && + action.message is ServerMessage.SessionSnapshot && + action.message.sessionId.value != state.sessions.selectedId + ) Action.NoOp else action + + log.debug("input state before reducers={}", state.input) val (afterInput, ie) = InputReducer.reduce(state, action) + log.debug("sessions state before reducers={}", state.sessions) val (sessions, se) = SessionsReducer.reduce(afterInput.sessions, prevInputMode, prevInputBuffer, action, clock) - val (approval, ae) = ApprovalReducer.reduce(afterInput.approval, prevInputMode, action) + log.debug("approval state before reducers={}", state.approval) + val (approval, ae) = ApprovalReducer.reduce(afterInput.approval, prevInputMode, approvalAction) + log.debug("connection state before reducers={}", state.connection) val (connection, ce) = ConnectionReducer.reduce(afterInput.connection, action) + log.debug("provider state before reducers={}", state.provider) val (provider, pe) = ProviderReducer.reduce(afterInput.provider, action) val approvalJustArrived = approval.active != null && afterInput.approval.active == null val withSubReducers = afterInput.copy( @@ -30,10 +46,12 @@ object RootReducer { approvalOverlayVisible = !withSubReducers.approvalOverlayVisible, eventOverlayVisible = false, ) + is Action.ToggleEventOverlay -> withSubReducers.copy( eventOverlayVisible = !withSubReducers.eventOverlayVisible, approvalOverlayVisible = false, ) + else -> withSubReducers } return final to (quitEffects + ie + se + ae + ce + pe) diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt index 3b1b1bf1..35b091ad 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt @@ -11,10 +11,13 @@ import com.correx.apps.tui.state.ToolDisplayStatus import com.correx.apps.tui.state.TuiEventEntry import com.correx.apps.tui.state.TuiToolRecord import com.correx.core.events.types.SessionId +import org.slf4j.LoggerFactory import java.time.Instant import java.time.ZoneOffset import java.time.format.DateTimeFormatter +private val log = LoggerFactory.getLogger(SessionsReducer::class.simpleName) + object SessionsReducer { private val timeFormatter = DateTimeFormatter.ofPattern("HH:mm:ss").withZone(ZoneOffset.UTC) @@ -103,7 +106,32 @@ object SessionsReducer { is ServerMessage.ToolCompleted -> processToolCompletedMessage(msg, sessions) is ServerMessage.ToolFailed -> processToolFailedMessage(msg, sessions) is ServerMessage.ToolRejected -> processToolRejectedMessage(clock, sessions, msg) + is ServerMessage.SessionSnapshot -> processSessionSnapshotMessage(clock, sessions, msg) else -> sessions to emptyList() + }.also { log.debug("processed server message: {}", msg) } + + private fun processSessionSnapshotMessage( + clock: () -> Long, + sessionState: SessionsState, + msg: ServerMessage.SessionSnapshot, + ): Pair> { + val summary = SessionSummary( + id = msg.sessionId.value, + status = when { + msg.pendingApproval -> "PAUSED awaiting approval" + else -> msg.status + }, + workflowId = msg.workflowId, + name = msg.workflowId, + lastEventAt = clock(), + currentStage = msg.currentStageId, + currentStageId = msg.currentStageId, + ) + val selected = sessionState.selectedId ?: msg.sessionId.value + return sessionState.copy( + sessions = sessionState.sessions + summary, + selectedId = selected, + ) to listOf(Effect.ConnectSession(msg.sessionId)) } private fun processToolRejectedMessage( diff --git a/core/approvals/src/main/kotlin/com/correx/core/approvals/DefaultApprovalReducer.kt b/core/approvals/src/main/kotlin/com/correx/core/approvals/DefaultApprovalReducer.kt index fa931292..5be1e9ef 100644 --- a/core/approvals/src/main/kotlin/com/correx/core/approvals/DefaultApprovalReducer.kt +++ b/core/approvals/src/main/kotlin/com/correx/core/approvals/DefaultApprovalReducer.kt @@ -26,7 +26,9 @@ class DefaultApprovalReducer : ApprovalReducer { timestamp = event.metadata.timestamp, causationId = event.metadata.causationId, correlationId = event.metadata.correlationId, - userSteering = payload.userSteering + userSteering = payload.userSteering, + toolName = payload.toolName, + preview = payload.preview, ) state.copy(requests = state.requests + (request.id to request)) } @@ -39,11 +41,11 @@ class DefaultApprovalReducer : ApprovalReducer { identity = ApprovalScopeIdentity( sessionId = event.metadata.sessionId, stageId = null, - projectId = null + projectId = null, ), mode = ApprovalMode.PROMPT, causationId = event.metadata.causationId, - correlationId = event.metadata.correlationId + correlationId = event.metadata.correlationId, ) val decision = ApprovalDecision( id = decisionId, @@ -54,7 +56,7 @@ class DefaultApprovalReducer : ApprovalReducer { contextSnapshot = contextSnapshot, resolutionTimestamp = payload.resolutionTimestamp, reason = payload.reason, - userSteering = payload.userSteering + userSteering = payload.userSteering, ) state.copy(decisions = state.decisions + (decisionId to decision)) } @@ -66,7 +68,7 @@ class DefaultApprovalReducer : ApprovalReducer { permittedTiers = payload.permittedTiers, reason = payload.reason, timestamp = event.metadata.timestamp, - expiresAt = payload.expiresAt + expiresAt = payload.expiresAt, ) state.copy(grants = state.grants + (grant.id to grant)) } diff --git a/core/approvals/src/main/kotlin/com/correx/core/approvals/model/DomainApprovalRequest.kt b/core/approvals/src/main/kotlin/com/correx/core/approvals/model/DomainApprovalRequest.kt index 922095f0..7d2485db 100644 --- a/core/approvals/src/main/kotlin/com/correx/core/approvals/model/DomainApprovalRequest.kt +++ b/core/approvals/src/main/kotlin/com/correx/core/approvals/model/DomainApprovalRequest.kt @@ -19,5 +19,7 @@ data class DomainApprovalRequest( val timestamp: Instant, val causationId: CausationId? = null, val correlationId: CorrelationId? = null, - val userSteering: UserSteering? = null + val userSteering: UserSteering? = null, + val toolName: String? = null, + val preview: String? = null, ) diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/ApprovalEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/ApprovalEvents.kt index 8223d9d6..2138ea4a 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/ApprovalEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/ApprovalEvents.kt @@ -25,7 +25,9 @@ data class ApprovalRequestedEvent( val sessionId: SessionId, val stageId: StageId?, val projectId: ProjectId?, - val userSteering: UserSteering? = null + val userSteering: UserSteering? = null, + val toolName: String? = null, + val preview: String? = null, ) : EventPayload @Serializable @@ -37,7 +39,7 @@ data class ApprovalDecisionResolvedEvent( val tier: Tier, val resolutionTimestamp: Instant, val reason: String?, - val userSteering: UserSteering? = null + val userSteering: UserSteering? = null, ) : EventPayload @Serializable @@ -49,10 +51,10 @@ data class ApprovalGrantCreatedEvent( val expiresAt: Instant?, val sessionId: SessionId, val stageId: StageId?, - val projectId: ProjectId? + val projectId: ProjectId?, ) : EventPayload @Serializable data class ApprovalGrantExpiredEvent( - val grantId: GrantId + val grantId: GrantId, ) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt index 46a2bbf2..64958469 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/OrchestrationEvents.kt @@ -18,7 +18,6 @@ data class WorkflowCompletedEvent( val sessionId: SessionId, val terminalStageId: StageId, val totalStages: Int, - val workflowId: String, ) : EventPayload @Serializable diff --git a/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt b/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt index 32b17b18..c3274970 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationState.kt @@ -6,6 +6,7 @@ import kotlinx.serialization.Serializable @Serializable data class OrchestrationState( + val workflowId: String = "", val currentStageId: StageId? = null, val status: OrchestrationStatus = OrchestrationStatus.IDLE, val retryCount: Int = 0, diff --git a/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt b/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt index c20bef02..73ae179a 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt @@ -47,4 +47,10 @@ interface EventStore { * Intended for batch jobs (e.g. CAS liveness scans). Implementations should stream lazily. */ fun allEvents(): Sequence + + /** + * Return all known session ids in the store. + * Implementations should derive this from the session index, not by scanning allEvents(). + */ + fun allSessionIds(): Set } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt index dbd9339b..2eabbb1a 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt @@ -16,16 +16,22 @@ class DefaultOrchestrationReducer : OrchestrationReducer { event: StoredEvent, ): OrchestrationState = when (val p = event.payload) { is WorkflowStartedEvent -> state.copy( + workflowId = p.workflowId, status = OrchestrationStatus.RUNNING, currentStageId = p.startStageId, retryPolicy = p.retryPolicy, ) + is WorkflowCompletedEvent -> state.copy( status = OrchestrationStatus.COMPLETED, currentStageId = p.terminalStageId, ) - is WorkflowFailedEvent -> state.copy(status = OrchestrationStatus.FAILED, failureReason = p.reason) + is WorkflowFailedEvent -> state.copy( + status = OrchestrationStatus.FAILED, + failureReason = p.reason, + ) + is OrchestrationPausedEvent -> state.copy( status = OrchestrationStatus.PAUSED, pendingApproval = true, diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 11314037..ea410489 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -94,7 +94,7 @@ class DefaultSessionOrchestrator( } is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) { - completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount, enriched.graph.id) + completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount) } else { failWorkflow( sessionId = enriched.sessionId, @@ -140,7 +140,7 @@ class DefaultSessionOrchestrator( if (isTerminal(ctx.graph, nextStageId)) { // Terminal stage is a sentinel — not executed, so don't count it. return StepResult.Terminal( - completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id), + completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount), ) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt index 56d4c8fc..30384b22 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt @@ -1,5 +1,6 @@ package com.correx.core.kernel.orchestration +import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifacts.repository.ArtifactRepository import com.correx.core.events.stores.EventStore import com.correx.core.inference.InferenceRepository @@ -11,4 +12,5 @@ data class OrchestratorRepositories( val orchestrationRepository: OrchestrationRepository, val sessionRepository: DefaultSessionRepository, val artifactRepository: ArtifactRepository, + val approvalRepository: DefaultApprovalRepository, ) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt index df115adc..18e3ced7 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt @@ -87,7 +87,7 @@ class ReplayOrchestrator( val nextStageId = decision.to if (isTerminal(ctx.graph, nextStageId)) { - return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id) + return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount) } when (val result = enterStage(ctx, nextStageId, session)) { @@ -101,7 +101,7 @@ class ReplayOrchestrator( } is TransitionDecision.Stay -> if (isTerminal(ctx.graph, ctx.currentStageId)) { - completeWorkflow(ctx.sessionId, ctx.currentStageId, ctx.stageCount, ctx.graph.id) + completeWorkflow(ctx.sessionId, ctx.currentStageId, ctx.stageCount) } else { step(ctx) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 5f9e75b9..28bea902 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -112,7 +112,6 @@ abstract class SessionOrchestrator( private val toolExecutor: ToolExecutor? = engines.toolExecutor private val toolRegistry: ToolRegistry? = engines.toolRegistry private val inferenceRepository: InferenceRepository = repositories.inferenceRepository - private val artifactRepository = repositories.artifactRepository internal val orchestrationRepository: OrchestrationRepository = repositories.orchestrationRepository internal abstract val cancellations: ConcurrentHashMap internal val pendingApprovals: ConcurrentHashMap> = @@ -145,7 +144,7 @@ abstract class SessionOrchestrator( .onFailure { log.error( "[SessionOrchestrator] stage=${stageId.value}: " + - "failed to load prompt '$path': ${it.message}", + "failed to load prompt '$path': ${it.message}", ) } .getOrNull() @@ -168,7 +167,7 @@ abstract class SessionOrchestrator( .onFailure { log.error( "[SessionOrchestrator] stage=${stageId.value}: " + - "failed to load default system prompt '$path': ${it.message}", + "failed to load default system prompt '$path': ${it.message}", ) } .getOrNull() @@ -192,7 +191,7 @@ abstract class SessionOrchestrator( .onFailure { log.error( "[SessionOrchestrator] stage=${stageId.value}: " + - "failed to load prompt '$path': ${it.message}", + "failed to load prompt '$path': ${it.message}", ) } .getOrNull() @@ -214,7 +213,7 @@ abstract class SessionOrchestrator( val llmEmittedSlots = stageConfig.produces.filter { it.kind.llmEmitted } require(llmEmittedSlots.size <= 1) { "Stage '${stageId.value}' declares ${llmEmittedSlots.size} LLM-emitted artifact kinds; " + - "only 0 or 1 is supported" + "only 0 or 1 is supported" } val responseFormat = llmEmittedSlots.firstOrNull() ?.let { ResponseFormat.Json(it.kind.deriveJsonSchema()) } @@ -269,6 +268,7 @@ abstract class SessionOrchestrator( emitToolArtifacts(sessionId, stageId, stageConfig) verifyProduces(sessionId, stageId, stageConfig) } + is StageExecutionResult.Failure -> outcome } } @@ -321,6 +321,8 @@ abstract class SessionOrchestrator( validationReportId = ValidationReportId(UUID.randomUUID().toString()), riskSummaryId = null, timestamp = Clock.System.now(), + toolName = toolCall.function.name, + preview = toolCall.function.arguments.take(200), ) emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING")) emit( @@ -333,6 +335,8 @@ abstract class SessionOrchestrator( sessionId = sessionId, stageId = stageId, projectId = null, + toolName = toolCall.function.name, + preview = toolCall.function.arguments.take(200), ), ) val deferred = CompletableDeferred() @@ -399,8 +403,8 @@ abstract class SessionOrchestrator( if (responseFormat !is ResponseFormat.Json) return emptyList() val compactSchema = Json.encodeToString(JsonSchema.serializer(), responseFormat.schema) val instruction = "Respond with a single JSON object matching this schema. " + - "Do not include markdown, code fences, or commentary outside the JSON. " + - "Schema: $compactSchema" + "Do not include markdown, code fences, or commentary outside the JSON. " + + "Schema: $compactSchema" return listOf( ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), @@ -442,7 +446,7 @@ abstract class SessionOrchestrator( val missingIds = missing.joinToString(", ") { it.value } log.error( "[Orchestrator] stage missing produced artifacts " + - "session=${sessionId.value} stage=${stageId.value} missing=$missingIds", + "session=${sessionId.value} stage=${stageId.value} missing=$missingIds", ) return StageExecutionResult.Failure( "stage ${stageId.value} did not produce declared artifacts: $missingIds", @@ -598,13 +602,12 @@ abstract class SessionOrchestrator( sessionId: SessionId, terminalStageId: StageId, stageCount: Int, - workflowId: String, ): WorkflowResult.Completed { log.debug( "[Orchestrator] COMPLETED session={} terminalStage={} stages={}", sessionId.value, terminalStageId.value, stageCount, ) - emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId)) + emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount)) cancellations.remove(sessionId) return WorkflowResult.Completed(sessionId, terminalStageId) } @@ -642,6 +645,7 @@ abstract class SessionOrchestrator( // --- event emission --- internal suspend fun emit(sessionId: SessionId, payload: EventPayload) { + log.debug("[session {}] emitting event, payload: {}", sessionId.value.substring(0, 7), payload) eventStore.append( NewEvent( metadata = EventMetadata( diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt index 9a19f204..041c2102 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt @@ -63,6 +63,8 @@ class InMemoryEventStore : EventStore { .flatMap { stream -> synchronized(stream) { stream.toList() } } .asSequence() + override fun allSessionIds(): Set = sequences.keys + private fun doAppend(event: NewEvent, stream: MutableList): StoredEvent { val seq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) } .incrementAndGet() diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt index 65ce7c22..360cbdc1 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt @@ -42,7 +42,7 @@ class SqliteEventStore( correlation_id TEXT, payload TEXT NOT NULL ); - """.trimIndent() + """.trimIndent(), ) } } @@ -60,7 +60,7 @@ class SqliteEventStore( val s = StoredEvent( metadata = event.metadata, sequence = seq, - payload = event.payload + payload = event.payload, ) insert(s) @@ -91,7 +91,7 @@ class SqliteEventStore( val s = StoredEvent( metadata = event.metadata, sequence = seq, - payload = event.payload + payload = event.payload, ) insert(s) @@ -111,7 +111,7 @@ class SqliteEventStore( SELECT * FROM events WHERE session_id = ? ORDER BY sequence ASC - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.setString(1, sessionId.value) @@ -131,7 +131,7 @@ class SqliteEventStore( WHERE session_id = ? AND sequence > ? ORDER BY sequence ASC - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.setString(1, sessionId.value) ps.setLong(2, fromSequence) @@ -151,7 +151,7 @@ class SqliteEventStore( SELECT MAX(sequence) FROM events WHERE session_id = ? - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.setString(1, sessionId.value) @@ -169,7 +169,7 @@ class SqliteEventStore( """ SELECT * FROM events ORDER BY session_id ASC, sequence ASC - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.executeQuery().use { rs -> buildList { @@ -180,6 +180,15 @@ class SqliteEventStore( } }.asSequence() + override fun allSessionIds(): Set = + mutableSetOf().apply { + connection.prepareStatement("SELECT DISTINCT session_id FROM events").use { ps -> + ps.executeQuery().use { rs -> + while (rs.next()) add(SessionId(rs.getString("session_id"))) + } + } + } + // ---------- helpers ---------- private fun nextSequence(sessionId: SessionId): Long = @@ -188,7 +197,7 @@ class SqliteEventStore( SELECT COALESCE(MAX(sequence), 0) FROM events WHERE session_id = ? - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.setString(1, sessionId.value) ps.executeQuery().use { rs -> @@ -211,7 +220,7 @@ class SqliteEventStore( correlation_id, payload ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.setString(1, event.metadata.eventId.value) ps.setString(2, event.metadata.sessionId.value) @@ -231,7 +240,7 @@ class SqliteEventStore( """ SELECT * FROM events WHERE event_id = ? - """.trimIndent() + """.trimIndent(), ).use { ps -> ps.setString(1, eventId.value) @@ -250,8 +259,8 @@ private fun ResultSet.toStoredEvent(jsonSerializer: JsonEventSerializer): Stored timestamp = Instant.parse(getString("timestamp")), schemaVersion = getInt("schema_version"), causationId = getString("causation_id")?.let { CausationId(it) }, - correlationId = getString("correlation_id")?.let { CorrelationId(it) } + correlationId = getString("correlation_id")?.let { CorrelationId(it) }, ), sequence = getLong("sequence"), - payload = jsonSerializer.deserialize(getString("payload")) + payload = jsonSerializer.deserialize(getString("payload")), ) diff --git a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt index 75aa9f12..f6c8a052 100644 --- a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt +++ b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt @@ -1,15 +1,15 @@ package com.correx.infrastructure +import com.correx.core.approvals.ApprovalProjector +import com.correx.core.approvals.DefaultApprovalReducer +import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifacts.DefaultArtifactReducer import com.correx.core.artifacts.repository.ArtifactRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.EventDispatcher import com.correx.core.events.stores.EventStore -import com.correx.core.inference.DefaultInferenceRouter import com.correx.core.inference.InferenceProvider import com.correx.core.inference.InferenceRouter -import com.correx.core.inference.ProviderRegistry -import com.correx.core.inference.RoutingStrategy import com.correx.core.router.DefaultRouterContextBuilder import com.correx.core.router.DefaultRouterFacade import com.correx.core.router.DefaultRouterReducer @@ -24,11 +24,8 @@ import com.correx.infrastructure.artifactscas.CasArtifactStore import com.correx.infrastructure.artifactscas.config.CasConfig import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex import com.correx.infrastructure.inference.DefaultProviderRegistry -import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy import com.correx.infrastructure.inference.commons.ModelDescriptor -import com.correx.infrastructure.inference.commons.ModelManager import com.correx.infrastructure.inference.commons.ResidencyMode -import com.correx.infrastructure.inference.llama.cpp.DefaultModelManager import com.correx.infrastructure.inference.llama.cpp.LlamaCppInferenceProvider import com.correx.infrastructure.persistence.SqliteEventStore import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository @@ -41,7 +38,6 @@ import com.correx.infrastructure.workflow.FileSystemPromptLoader import com.correx.infrastructure.workflow.PromptLoader import com.correx.infrastructure.workflow.TomlWorkflowLoader import com.correx.infrastructure.workflow.WorkflowLoader -import io.ktor.client.HttpClient import kotlinx.coroutines.runBlocking import java.nio.file.Files import java.nio.file.Path @@ -71,25 +67,9 @@ object InfrastructureModule { return runBlocking { CasArtifactStore.open(CasConfig(rootDir), index) } } - fun createModelManager( - eventStore: EventStore, - httpClient: HttpClient, - eventDispatcher: EventDispatcher, - ): ModelManager = DefaultModelManager( - eventStore = eventStore, - httpClient = httpClient, - eventDispatcher = eventDispatcher, - ) - fun createToolRegistry(config: ToolConfig): ToolRegistry = DefaultToolRegistry.build(config.buildTools()) - fun createInferenceRouter( - providers: List = emptyList(), - registry: ProviderRegistry = DefaultProviderRegistry(providers), - strategy: RoutingStrategy = FirstAvailableRoutingStrategy(), - ): InferenceRouter = DefaultInferenceRouter(registry, strategy) - fun createLlamaCppProvider( modelId: String, modelPath: String, @@ -111,6 +91,16 @@ object InfrastructureModule { fun createArtifactRepository(eventStore: EventStore): ArtifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()) + fun createApprovalRepository(eventStore: EventStore): DefaultApprovalRepository = + DefaultApprovalRepository( + DefaultEventReplayer( + eventStore, + ApprovalProjector( + DefaultApprovalReducer(), + ), + ), + ) + fun createWorkflowLoader(): WorkflowLoader = TomlWorkflowLoader() fun createPromptLoader(): PromptLoader = FileSystemPromptLoader() diff --git a/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileEditTool.kt b/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileEditTool.kt index 5c3dea24..ecc7fd1f 100644 --- a/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileEditTool.kt +++ b/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileEditTool.kt @@ -13,7 +13,11 @@ import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonArray import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonObject import java.io.IOException import java.nio.file.Files import java.nio.file.InvalidPathException @@ -39,7 +43,35 @@ class FileEditTool( override val name: String = "file_edit" override val description: String = "Edit content in a file at the specified path" - override val parametersSchema: JsonObject = buildJsonObject {} + override val parametersSchema: JsonObject = buildJsonObject { + put("type", "object") + putJsonObject("properties") { + putJsonObject("path") { + put("type", "string") + put("description", "Relative path to the file to edit") + } + putJsonObject("operation") { + put("type", "string") + put("description", "Either 'append', 'replace', or 'patch'") + } + putJsonObject("content") { + put("type", "string") + put("description", "Content to write. Used for 'append' and 'replace'.") + } + putJsonObject("old_str") { + put("type", "string") + put("description", "Exact string to find. Required for 'patch'.") + } + putJsonObject("new_str") { + put("type", "string") + put("description", "Replacement string. Required for 'patch'.") + } + } + put("required", buildJsonArray { + add(JsonPrimitive("path")) + add(JsonPrimitive("operation")) + }) + } override val tier: Tier = Tier.T3 override val requiredCapabilities: Set = setOf(ToolCapability.FILE_WRITE) diff --git a/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileReadTool.kt b/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileReadTool.kt index fe8688e7..b0a70c9a 100644 --- a/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileReadTool.kt +++ b/infrastructure/tools/filesystem/src/main/kotlin/com/correx/infrastructure/tools/filesystem/FileReadTool.kt @@ -7,11 +7,14 @@ import com.correx.core.tools.contract.ToolCapability import com.correx.core.tools.contract.ToolExecutor import com.correx.core.tools.contract.ToolResult import com.correx.core.tools.contract.ValidationResult -import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonArray import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonObject import java.io.IOException import java.nio.file.Files import java.nio.file.InvalidPathException @@ -24,76 +27,92 @@ class FileReadTool( override val name: String = "file_read" override val description: String = "Read content from a file at the specified path" - override val parametersSchema: JsonObject = buildJsonObject {} + override val parametersSchema: JsonObject = buildJsonObject { + put("type", "object") + putJsonObject("properties") { + putJsonObject("path") { + put("type", "string") + put("description", "Relative path to the file to read") + } + putJsonObject("start_line") { + put("type", "integer") + put("description", "First line to read, 1-indexed inclusive. Omit to read from beginning.") + } + putJsonObject("end_line") { + put("type", "integer") + put("description", "Last line to read, 1-indexed inclusive. Omit to read to end of file.") + } + } + put("required", buildJsonArray { add(JsonPrimitive("path")) }) + } override val tier: Tier = Tier.T1 override val requiredCapabilities: Set = setOf(ToolCapability.FILE_READ) - override fun validateRequest(request: ToolRequest): ValidationResult = - (request.parameters["path"] as? String)?.let { pathString -> - runCatching { - val path = Paths.get(pathString).normalize().toAbsolutePath() - if (allowedPaths.none { path.startsWith(it.normalize().toAbsolutePath()) }) { - return@runCatching ValidationResult.Invalid("Path '$pathString' is not in the allowed list") - } + override fun validateRequest(request: ToolRequest): ValidationResult { + val pathString = request.parameters["path"] as? String + ?: return ValidationResult.Invalid("Missing or invalid 'path' parameter. Expected String.") + val startLine = request.parameters["start_line"]?.toString()?.toIntOrNull() + val endLine = request.parameters["end_line"]?.toString()?.toIntOrNull() + + if (startLine != null && startLine < 1) + return ValidationResult.Invalid("'start_line' must be >= 1.") + if (endLine != null && endLine < 1) + return ValidationResult.Invalid("'end_line' must be >= 1.") + if (startLine != null && endLine != null && endLine < startLine) + return ValidationResult.Invalid("'end_line' must be >= 'start_line'.") + + return runCatching { + val path = Paths.get(pathString).normalize().toAbsolutePath() + if (allowedPaths.isNotEmpty() && allowedPaths.none { path.startsWith(it.normalize().toAbsolutePath()) }) + ValidationResult.Invalid("Path '$pathString' is not in the allowed list") + else ValidationResult.Valid - }.getOrElse { - when (it) { - is InvalidPathException -> ValidationResult.Invalid("Invalid path format: ${it.message}") - is IOException -> ValidationResult.Invalid("IO error: ${it.message}") - is SecurityException -> ValidationResult.Invalid("Security error: ${it.message}") - else -> ValidationResult.Invalid(it.message ?: "Unknown error occured") - } + }.getOrElse { + when (it) { + is InvalidPathException -> ValidationResult.Invalid("Invalid path format: ${it.message}") + is IOException -> ValidationResult.Invalid("IO error: ${it.message}") + is SecurityException -> ValidationResult.Invalid("Security error: ${it.message}") + else -> ValidationResult.Invalid(it.message ?: "Unknown error occurred") } - } ?: ValidationResult.Invalid("Missing or invalid 'path' parameter. Expected String.") + } + } override suspend fun execute(request: ToolRequest): ToolResult = withContext(Dispatchers.IO) { - validateRequest(request).run { - takeIf { it is ValidationResult.Valid }?.let { - val pathString = request.parameters["path"] as String - val path = Paths.get(pathString).normalize().toAbsolutePath() - if (Files.exists(path)) { - readStringCatching(request, path, pathString) - } else { - ToolResult.Failure( - invocationId = request.invocationId, - reason = "File not found: $pathString", - recoverable = false, - ) - } - } ?: ToolResult.Failure( + val validation = validateRequest(request) + if (validation is ValidationResult.Invalid) + return@withContext ToolResult.Failure( invocationId = request.invocationId, - reason = (this as ValidationResult.Invalid).reason, + reason = validation.reason, + recoverable = false, + ) + + val pathString = request.parameters["path"] as String + val path = Paths.get(pathString).normalize().toAbsolutePath() + val startLine = request.parameters["start_line"]?.toString()?.toIntOrNull() + val endLine = request.parameters["end_line"]?.toString()?.toIntOrNull() + + if (!Files.exists(path)) + return@withContext ToolResult.Failure( + invocationId = request.invocationId, + reason = "File not found: $pathString", + recoverable = false, + ) + + runCatching { + val lines = Files.readAllLines(path) + val start = ((startLine ?: 1) - 1).coerceAtLeast(0) + val end = (endLine ?: lines.size).coerceAtMost(lines.size) + val content = lines.subList(start, end).joinToString("\n") + ToolResult.Success( + invocationId = request.invocationId, + output = content, + ) + }.getOrElse { + ToolResult.Failure( + invocationId = request.invocationId, + reason = "Failed to read file: ${it.message}", recoverable = false, ) } } - - private fun readStringCatching(request: ToolRequest, path: Path, pathString: String): ToolResult = - runCatching { - ToolResult.Success( - invocationId = request.invocationId, - output = Files.readString(path), - ) - }.getOrElse { - when (it) { - is CancellationException -> throw it - is SecurityException -> ToolResult.Failure( - invocationId = request.invocationId, - reason = "Access denied: $pathString, ${it.message}", - recoverable = false, - ) - - is IOException -> ToolResult.Failure( - invocationId = request.invocationId, - reason = "IO error: ${it.message}", - recoverable = false, - ) - - else -> ToolResult.Failure( - invocationId = request.invocationId, - reason = it.message ?: "Unknown error occurred while reading file", - recoverable = false, - ) - } - } } diff --git a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/shell/ShellTool.kt b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/shell/ShellTool.kt index c08ca07d..045ff06b 100644 --- a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/shell/ShellTool.kt +++ b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/shell/ShellTool.kt @@ -12,8 +12,13 @@ import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.async import kotlinx.coroutines.withContext import kotlinx.coroutines.withTimeout +import kotlinx.serialization.json.Json import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.buildJsonArray import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonObject import java.nio.file.Path class ShellTool( @@ -23,18 +28,36 @@ class ShellTool( ) : Tool, ToolExecutor { override val name: String = "shell" override val description: String = "Execute a shell command" - override val parametersSchema: JsonObject = buildJsonObject {} + override val parametersSchema: JsonObject = buildJsonObject { + put("type", "object") + putJsonObject("properties") { + putJsonObject("argv") { + put("type", "array") + putJsonObject("items") { put("type", "string") } + put("description", "Command and arguments as a list. Example: [\"bash\", \"script.sh\"]") + } + } + put("required", buildJsonArray { add(JsonPrimitive("argv")) }) + } override val tier: Tier = Tier.T2 override val requiredCapabilities: Set = setOf(ToolCapability.SHELL_EXEC, ToolCapability.PROCESS_SPAWN) override fun validateRequest(request: ToolRequest): ValidationResult { - val argv = (request.parameters["argv"] as? List<*>)?.filterIsInstance() + val argv = when (val raw = request.parameters["argv"]) { + is List<*> -> raw.filterIsInstance() + is String -> runCatching { + Json.decodeFromString>(raw) + }.getOrElse { emptyList() } + else -> emptyList() + } return when { - argv.isNullOrEmpty() -> + argv.isEmpty() -> ValidationResult.Invalid("Missing or empty 'argv' parameter. Expected List.") + allowedExecutables.isNotEmpty() && argv[0] !in allowedExecutables -> ValidationResult.Invalid("Executable '${argv[0]}' is not in the allowed list.") + else -> ValidationResult.Valid } } @@ -42,7 +65,13 @@ class ShellTool( override suspend fun execute(request: ToolRequest): ToolResult = withContext(Dispatchers.IO) { validateRequest(request).run { takeIf { this is ValidationResult.Valid }?.let { - val argv = (request.parameters["argv"] as? List<*>)?.filterIsInstance() + val argv = when (val raw = request.parameters["argv"]) { + is List<*> -> raw.filterIsInstance() + is String -> runCatching { + Json.decodeFromString>(raw) + }.getOrElse { emptyList() } + else -> emptyList() + } val process = ProcessBuilder(argv).apply { workingDir?.let { directory(it.toFile()) } }.start() diff --git a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/orchestration/EventsTest.kt b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/orchestration/EventsTest.kt index 0edb62a2..4491d995 100644 --- a/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/orchestration/EventsTest.kt +++ b/testing/contracts/src/test/kotlin/com/correx/testing/contracts/model/orchestration/EventsTest.kt @@ -37,7 +37,6 @@ class EventsTest { sessionId = SessionId("s1"), terminalStageId = StageId("stage-a"), totalStages = 1, - workflowId = workflowId, ) val json = eventJson.encodeToString(EventPayload.serializer(), event) diff --git a/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt b/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt index 1ba69c69..d9625afe 100644 --- a/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt +++ b/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt @@ -5,14 +5,20 @@ import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.stores.EventStore import com.correx.core.events.types.ContextPackId +import com.correx.core.events.types.EventId import com.correx.core.events.types.InferenceRequestId +import com.correx.core.events.types.ProviderId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId +import com.correx.core.inference.CapabilityScore +import com.correx.core.inference.FinishReason import com.correx.core.inference.InferenceProvider import com.correx.core.inference.InferenceRequest import com.correx.core.inference.InferenceResponse import com.correx.core.inference.InferenceRouter import com.correx.core.inference.ModelCapability +import com.correx.core.inference.ProviderHealth +import com.correx.core.inference.ResponseFormat import com.correx.core.inference.TokenUsage import com.correx.core.router.ChatMode import com.correx.core.router.DefaultRouterFacade @@ -24,6 +30,8 @@ import com.correx.core.router.model.RouterResponse import com.correx.core.router.model.RouterState import com.correx.core.router.model.TurnRole import com.correx.core.router.model.WorkflowStatus +import com.correx.testing.fixtures.inference.MockTokenizer +import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertFalse @@ -405,25 +413,25 @@ class RouterFacadeTest { requiredCapabilities: Set, ): InferenceProvider { return object : InferenceProvider { - override val id = com.correx.core.events.types.ProviderId("mock") + override val id = ProviderId("mock") override val name = "Mock" - override val tokenizer = com.correx.testing.fixtures.inference.MockTokenizer() + override val tokenizer = MockTokenizer() override suspend fun infer(request: InferenceRequest): InferenceResponse { assertEquals(1, capturedContextPacks.size) assertEquals(capturedContextPacks[0].id, request.contextPack.id) return InferenceResponse( requestId = request.requestId, text = "response", - finishReason = com.correx.core.inference.FinishReason.Stop, + finishReason = FinishReason.Stop, tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5), latencyMs = 0, ) } - override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth = - com.correx.core.inference.ProviderHealth.Healthy + override suspend fun healthCheck(): ProviderHealth = + ProviderHealth.Healthy - override fun capabilities(): Set = emptySet() + override fun capabilities(): Set = emptySet() } } }, @@ -456,7 +464,7 @@ class RouterFacadeTest { ) facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!") val req = capturedRequests[0] - assertTrue(req.responseFormat is com.correx.core.inference.ResponseFormat.Text) + assertTrue(req.responseFormat is ResponseFormat.Text) } @Test @@ -512,22 +520,22 @@ class RouterFacadeTest { private fun mockProvider(responseText: String): InferenceProvider = object : InferenceProvider { - override val id = com.correx.core.events.types.ProviderId("mock") + override val id = ProviderId("mock") override val name = "Mock" - override val tokenizer = com.correx.testing.fixtures.inference.MockTokenizer() + override val tokenizer = MockTokenizer() override suspend fun infer(request: InferenceRequest): InferenceResponse = InferenceResponse( requestId = request.requestId, text = responseText, - finishReason = com.correx.core.inference.FinishReason.Stop, + finishReason = FinishReason.Stop, tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5), latencyMs = 0, ) - override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth = - com.correx.core.inference.ProviderHealth.Healthy + override suspend fun healthCheck(): ProviderHealth = + ProviderHealth.Healthy - override fun capabilities(): Set = - setOf(com.correx.core.inference.CapabilityScore(ModelCapability.General, 1.0)) + override fun capabilities(): Set = + setOf(CapabilityScore(ModelCapability.General, 1.0)) } private fun mockProviderWithCapture( @@ -535,24 +543,24 @@ class RouterFacadeTest { requestIds: MutableList, ): InferenceProvider = object : InferenceProvider { - override val id = com.correx.core.events.types.ProviderId("mock") + override val id = ProviderId("mock") override val name = "Mock" - override val tokenizer = com.correx.testing.fixtures.inference.MockTokenizer() + override val tokenizer = MockTokenizer() override suspend fun infer(request: InferenceRequest): InferenceResponse { requestIds.add(request.requestId) return InferenceResponse( requestId = request.requestId, text = responseText, - finishReason = com.correx.core.inference.FinishReason.Stop, + finishReason = FinishReason.Stop, tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5), latencyMs = 0, ) } - override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth = - com.correx.core.inference.ProviderHealth.Healthy + override suspend fun healthCheck(): ProviderHealth = + ProviderHealth.Healthy - override fun capabilities(): Set = emptySet() + override fun capabilities(): Set = emptySet() } private fun mockProviderWithRequestCapture( @@ -560,24 +568,24 @@ class RouterFacadeTest { requests: MutableList, ): InferenceProvider = object : InferenceProvider { - override val id = com.correx.core.events.types.ProviderId("mock") + override val id = ProviderId("mock") override val name = "Mock" - override val tokenizer = com.correx.testing.fixtures.inference.MockTokenizer() + override val tokenizer = MockTokenizer() override suspend fun infer(request: InferenceRequest): InferenceResponse { requests.add(request) return InferenceResponse( requestId = request.requestId, text = responseText, - finishReason = com.correx.core.inference.FinishReason.Stop, + finishReason = FinishReason.Stop, tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5), latencyMs = 0, ) } - override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth = - com.correx.core.inference.ProviderHealth.Healthy + override suspend fun healthCheck(): ProviderHealth = + ProviderHealth.Healthy - override fun capabilities(): Set = emptySet() + override fun capabilities(): Set = emptySet() } private fun emptyContextPack(): ContextPack = ContextPack( @@ -591,7 +599,7 @@ class RouterFacadeTest { private class MapBackedEventStore : EventStore { val appendedEvents = mutableListOf() - private val storedEvents = mutableMapOf() + private val storedEvents: MutableMap = mutableMapOf() private var nextSequence = 1L override suspend fun append(event: NewEvent): StoredEvent { @@ -620,10 +628,12 @@ class RouterFacadeTest { override fun lastSequence(sessionId: SessionId): Long? = read(sessionId).maxOfOrNull { it.sequence } - override fun subscribe(sessionId: SessionId): kotlinx.coroutines.flow.Flow = + override fun subscribe(sessionId: SessionId): Flow = throw UnsupportedOperationException("subscribe not implemented for mock") override fun allEvents(): Sequence = storedEvents.values.asSequence() + + override fun allSessionIds(): Set = storedEvents.values.map { it.metadata.sessionId }.toSet() } } diff --git a/testing/deterministic/src/test/kotlin/RouterProjectorTest.kt b/testing/deterministic/src/test/kotlin/RouterProjectorTest.kt index 4ba8ae1d..bc922b64 100644 --- a/testing/deterministic/src/test/kotlin/RouterProjectorTest.kt +++ b/testing/deterministic/src/test/kotlin/RouterProjectorTest.kt @@ -65,7 +65,7 @@ class RouterProjectorTest { ) val updated = projector.apply( started, - stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3)), ) assertEquals(WorkflowStatus.COMPLETED, updated.workflowStatus) assertNull(updated.currentStageId) @@ -176,7 +176,7 @@ class RouterProjectorTest { stored(payload = OrchestrationResumedEvent(sessionId, StageId("a"))), stored(payload = StageCompletedEvent(sessionId, StageId("b"), StageId("t2"))), stored(payload = StageFailedEvent(sessionId, StageId("b"), StageId("t3"), "error")), - stored(payload = WorkflowCompletedEvent(sessionId, StageId("c"), 3, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, StageId("c"), 3)), ) var projected = state @@ -206,7 +206,7 @@ class RouterProjectorTest { state = projector.apply( state, - stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1)), ) assertEquals(WorkflowStatus.COMPLETED, state.workflowStatus) assertNull(state.currentStageId) diff --git a/testing/deterministic/src/test/kotlin/RouterReducerTest.kt b/testing/deterministic/src/test/kotlin/RouterReducerTest.kt index cdca0736..dde79b61 100644 --- a/testing/deterministic/src/test/kotlin/RouterReducerTest.kt +++ b/testing/deterministic/src/test/kotlin/RouterReducerTest.kt @@ -58,7 +58,7 @@ class RouterReducerTest { ) val completed = reducer.reduce( started, - stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3)), ) assertEquals(WorkflowStatus.COMPLETED, completed.workflowStatus) assertNull(completed.currentStageId) @@ -233,7 +233,7 @@ class RouterReducerTest { assertEquals(StageOutcomeKind.SUCCESS, s2.l2Memory[0].outcome) val s3 = - reducer.reduce(s2, stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1, workflowId))) + reducer.reduce(s2, stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1))) assertEquals(WorkflowStatus.COMPLETED, s3.workflowStatus) assertNull(s3.currentStageId) diff --git a/testing/projections/src/test/kotlin/OrchestrationProjectorTest.kt b/testing/projections/src/test/kotlin/OrchestrationProjectorTest.kt index d301368e..8be46186 100644 --- a/testing/projections/src/test/kotlin/OrchestrationProjectorTest.kt +++ b/testing/projections/src/test/kotlin/OrchestrationProjectorTest.kt @@ -81,7 +81,7 @@ class OrchestrationProjectorTest { failureReason = "Something went wrong", ), ), - stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)), ) val result1 = events.fold(initialState) { state, event -> @@ -101,7 +101,7 @@ class OrchestrationProjectorTest { val s2 = projector.apply(s1, stored(payload = OrchestrationPausedEvent(sessionId, stageId, "approval required"))) val s3 = projector.apply(s2, stored(payload = OrchestrationResumedEvent(sessionId, stageId))) - val s4 = projector.apply(s3, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId))) + val s4 = projector.apply(s3, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1))) assertEquals(OrchestrationStatus.RUNNING, s1.status) assertEquals(OrchestrationStatus.PAUSED, s2.status) diff --git a/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt b/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt index 4e4023be..4bbc02bc 100644 --- a/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt +++ b/testing/projections/src/test/kotlin/OrchestrationReducerTest.kt @@ -21,8 +21,8 @@ class OrchestrationReducerTest { private val reducer = DefaultOrchestrationReducer() private val sessionId = SessionId("s1") private val stageId = StageId("stage-1") - private val state = OrchestrationState(stageId) private val workflowId = "workflow-test" + private val state = OrchestrationState(workflowId, stageId) @Test fun `WorkflowStartedEvent sets status to RUNNING`() { @@ -58,7 +58,7 @@ class OrchestrationReducerTest { val completed = reducer.reduce( started, - stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)), ) assertTrue(completed.failureReason.isNullOrBlank()) assertEquals(OrchestrationStatus.COMPLETED, completed.status) diff --git a/testing/projections/src/test/kotlin/RouterProjectorTest.kt b/testing/projections/src/test/kotlin/RouterProjectorTest.kt index 44aae4ae..75940ea4 100644 --- a/testing/projections/src/test/kotlin/RouterProjectorTest.kt +++ b/testing/projections/src/test/kotlin/RouterProjectorTest.kt @@ -49,7 +49,7 @@ class RouterProjectorTest { val started = projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId))) val completed = - projector.apply(started, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId))) + projector.apply(started, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1))) assertEquals(WorkflowStatus.COMPLETED, completed.workflowStatus) assertNull(completed.currentStageId) } @@ -134,7 +134,7 @@ class RouterProjectorTest { stored(payload = StageCompletedEvent(sessionId, StageId("stage-2"), StageId("t1"))), stored(payload = SteeringNoteAddedEvent(sessionId, "note")), stored(payload = StageFailedEvent(sessionId, stageId, StageId("t2"), "timeout")), - stored(payload = WorkflowCompletedEvent(sessionId, stageId, 2, workflowId)), + stored(payload = WorkflowCompletedEvent(sessionId, stageId, 2)), ) val result1 = events.fold(initialState) { state, event -> @@ -155,7 +155,7 @@ class RouterProjectorTest { projector.apply(s1, stored(payload = StageCompletedEvent(sessionId, StageId("stage-2"), StageId("t1")))) val s3 = projector.apply(s2, stored(payload = StageFailedEvent(sessionId, stageId, StageId("t2"), "timeout"))) val s4 = projector.apply(s3, stored(payload = WorkflowStartedEvent(sessionId, workflowId, StageId("stage-3")))) - val s5 = projector.apply(s4, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId))) + val s5 = projector.apply(s4, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1))) assertEquals(WorkflowStatus.RUNNING, s1.workflowStatus) assertEquals(StageId("stage-2"), s2.l2Memory[0].stageId)