diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt index fc2ff020..65cf6bfc 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt @@ -16,7 +16,7 @@ import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId -import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.kernel.orchestration.ApprovalGateway import com.correx.core.sessions.ApprovalMode import io.ktor.server.websocket.DefaultWebSocketServerSession import io.ktor.websocket.Frame @@ -28,7 +28,7 @@ import kotlinx.datetime.Clock import java.util.concurrent.ConcurrentHashMap class ApprovalCoordinator( - private val orchestrator: DefaultSessionOrchestrator, + private val orchestrator: ApprovalGateway, private val config: ApprovalConfig, private val scope: CoroutineScope, ) { diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/ApprovalRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/ApprovalRoutes.kt index 97edd8ac..c52a65d3 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/routes/ApprovalRoutes.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/ApprovalRoutes.kt @@ -1,8 +1,11 @@ +@file:Suppress("MatchingDeclarationName") + package com.correx.apps.server.routes import com.correx.apps.server.approval.ApprovalCoordinator import com.correx.apps.server.protocol.ApprovalDecision import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ServerMessage import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.SessionId import com.correx.core.utils.TypeId @@ -43,7 +46,8 @@ fun Route.approvalRoutes(coordinator: ApprovalCoordinator) { ) val error = coordinator.handleResponse(msg, sessionId) if (error != null) { - call.respond(HttpStatusCode.Conflict, mapOf("error" to (error as? com.correx.apps.server.protocol.ServerMessage.ProtocolError)?.message)) + val msg = (error as? ServerMessage.ProtocolError)?.message + call.respond(HttpStatusCode.Conflict, mapOf("error" to msg)) } else { call.respond(HttpStatusCode.OK) } diff --git a/core/events/build.gradle b/core/events/build.gradle index 228057df..97702810 100644 --- a/core/events/build.gradle +++ b/core/events/build.gradle @@ -6,4 +6,5 @@ plugins { dependencies { implementation "org.jetbrains.kotlinx:kotlinx-datetime" + implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version" } \ No newline at end of file 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 e6c15ff5..1dad28e8 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 @@ -3,6 +3,7 @@ package com.correx.core.events.stores import com.correx.core.events.events.NewEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.types.SessionId +import kotlinx.coroutines.flow.Flow interface EventStore { @@ -34,4 +35,10 @@ interface EventStore { * Get last sequence number for session. */ fun lastSequence(sessionId: SessionId): Long? + + /** + * Subscribe to live events for a session. Emits each event as it is appended. + * Does not replay historical events — use [readFrom] for catch-up before subscribing. + */ + fun subscribe(sessionId: SessionId): Flow } \ No newline at end of file diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt new file mode 100644 index 00000000..e9c13840 --- /dev/null +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt @@ -0,0 +1,8 @@ +package com.correx.core.kernel.orchestration + +import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.events.types.ApprovalRequestId + +interface ApprovalGateway { + fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) +} 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 8bd9ea53..38d4ceff 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 @@ -22,7 +22,7 @@ class DefaultSessionOrchestrator( private val repositories: OrchestratorRepositories, engines: OrchestratorEngines, private val retryCoordinator: RetryCoordinator, -) : SessionOrchestrator(repositories, engines) { +) : SessionOrchestrator(repositories, engines), ApprovalGateway { override val cancellations: ConcurrentHashMap = ConcurrentHashMap() @@ -82,7 +82,7 @@ class DefaultSessionOrchestrator( cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true) } - fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) { + override fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) { val deferred = pendingApprovals[requestId] ?: error("No pending approval for requestId ${requestId.value}") deferred.complete(decision) diff --git a/infrastructure/persistence/build.gradle b/infrastructure/persistence/build.gradle index 6c271c8c..67a575b7 100644 --- a/infrastructure/persistence/build.gradle +++ b/infrastructure/persistence/build.gradle @@ -5,6 +5,7 @@ plugins { } dependencies { + implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version" implementation(project(":core:events")) implementation(project(":core:sessions")) implementation "org.xerial:sqlite-jdbc" 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 f369a27e..c8240c84 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 @@ -5,6 +5,8 @@ import com.correx.core.events.events.StoredEvent import com.correx.core.events.stores.EventStore import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow import java.util.concurrent.* import java.util.concurrent.atomic.* @@ -12,6 +14,7 @@ class InMemoryEventStore : EventStore { private val streams = ConcurrentHashMap>() private val sequences = ConcurrentHashMap() private val seenEventIds = ConcurrentHashMap.newKeySet() + private val subscriptions = ConcurrentHashMap>() override fun append(event: NewEvent): StoredEvent { val stream = streams.computeIfAbsent(event.metadata.sessionId) { mutableListOf() } @@ -52,6 +55,9 @@ class InMemoryEventStore : EventStore { return sequences[sessionId]?.get() } + override fun subscribe(sessionId: SessionId): Flow = + subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) } + private fun doAppend(event: NewEvent, stream: MutableList): StoredEvent { val seq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) } .incrementAndGet() @@ -62,6 +68,7 @@ class InMemoryEventStore : EventStore { error("sequence violation for session ${event.metadata.sessionId}") } stream.add(stored) + subscriptions[event.metadata.sessionId]?.tryEmit(stored) return stored } 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 7357763d..63373936 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 @@ -11,14 +11,18 @@ import com.correx.core.events.types.CorrelationId import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.infrastructure.persistence.util.JDBCHelper.transaction +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.datetime.Instant import java.sql.Connection import java.sql.ResultSet +import java.util.concurrent.ConcurrentHashMap class SqliteEventStore( private val connection: Connection, private val jsonSerializer: JsonEventSerializer = JsonEventSerializer(eventJson) ) : EventStore { + private val subscriptions = ConcurrentHashMap>() init { connection.createStatement().use { stmt -> @@ -40,22 +44,23 @@ class SqliteEventStore( } override fun append(event: NewEvent): StoredEvent { - return connection.transaction { + val stored = connection.transaction { val existing = findByEventId(event.metadata.eventId) check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } val seq = nextSequence(event.metadata.sessionId) - val stored = StoredEvent( + val s = StoredEvent( metadata = event.metadata, sequence = seq, payload = event.payload ) - insert(stored) - - stored + insert(s) + s } + subscriptions[event.metadata.sessionId]?.tryEmit(stored) + return stored } override fun appendAll(events: List): List { @@ -63,24 +68,26 @@ class SqliteEventStore( val sessionId = events.first().metadata.sessionId - return connection.transaction { + val stored = connection.transaction { events.map { event -> val existing = findByEventId(event.metadata.eventId) check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } val seq = nextSequence(sessionId) - val stored = StoredEvent( + val s = StoredEvent( metadata = event.metadata, sequence = seq, payload = event.payload ) - insert(stored) - - stored + insert(s) + s } } + val flow = subscriptions[sessionId] + if (flow != null) stored.forEach { flow.tryEmit(it) } + return stored } override fun read(sessionId: SessionId): List = @@ -139,6 +146,9 @@ class SqliteEventStore( } } + override fun subscribe(sessionId: SessionId): Flow = + subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) } + // ---------- helpers ---------- private fun nextSequence(sessionId: SessionId): Long = diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index 8a995130..79aa8aac 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -1,8 +1,11 @@ -import com.correx.core.approvals.domain.ApprovalEngine +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.Tier import com.correx.core.approvals.domain.DefaultApprovalEngine import com.correx.core.approvals.model.ApprovalContext -import com.correx.core.approvals.model.ApprovalGrant -import com.correx.core.approvals.model.DomainApprovalRequest +import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.approvals.model.ApprovalScopeIdentity +import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.WorkflowCompletedEvent @@ -40,8 +43,11 @@ import com.correx.testing.fixtures.inference.MockInferenceProvider import com.correx.testing.fixtures.transitions.TransitionFixtures import com.correx.testing.fixtures.transitions.TransitionFixtures.threeStageGraph import com.correx.testing.kernel.MockSessionEventReplayer +import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking -import kotlinx.datetime.Instant +import kotlinx.coroutines.withTimeout +import kotlinx.coroutines.yield +import kotlinx.datetime.Clock import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertNotNull import org.junit.jupiter.api.Assertions.assertTrue @@ -166,52 +172,71 @@ class SessionOrchestratorIntegrationTest { retryCoordinator = retryCoordinator, ) - orchestrator.run(sessionId, graph, config) + val runJob = launch { orchestrator.run(sessionId, graph, config) } + + withTimeout(5_000) { + while (eventStore.read(sessionId).none { it.payload is OrchestrationPausedEvent }) { + yield() + } + } val events = eventStore.read(sessionId) val orPause = events.find { it.payload is OrchestrationPausedEvent } assertNotNull(orPause) val paused = orPause?.let { it.payload as? OrchestrationPausedEvent } assertEquals("APPROVAL_PENDING", paused?.reason) + + runJob.cancel() + runJob.join() } @Test fun `workflow resumes after approval`(): Unit = runBlocking { val sessionId = SessionId("s4") - val config = OrchestrationConfig( - retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0), - ) + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0)) val graph = threeStageGraph() - val yoloApprovalEngine = object : ApprovalEngine { - val inner = DefaultApprovalEngine() - override fun evaluate( - request: DomainApprovalRequest, - context: ApprovalContext, - grants: List, - now: Instant, - ) = - inner.evaluate(request, ApprovalContext(context.identity, ApprovalMode.YOLO), grants, now) - } - val approvingPipeline = ValidationPipeline( validators = listOf(cyclePolicyMissingValidator()), approvalTrigger = ApprovalTrigger(), ) - - val orchestrator = DefaultSessionOrchestrator( + val approvalOrchestrator = DefaultSessionOrchestrator( repositories = repositories, - engines = engines.copy( - validationPipeline = approvingPipeline, - approvalEngine = yoloApprovalEngine, - ), + engines = engines.copy(validationPipeline = approvingPipeline), retryCoordinator = retryCoordinator, ) - orchestrator.run(sessionId, graph, config) + val runJob = launch { approvalOrchestrator.run(sessionId, graph, config) } + + val requestId = withTimeout(5_000) { + var id: com.correx.core.events.types.ApprovalRequestId? = null + while (id == null) { + id = eventStore.read(sessionId) + .firstNotNullOfOrNull { it.payload as? ApprovalRequestedEvent } + ?.requestId + if (id == null) yield() + } + id + } + + val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = null, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val decision = ApprovalDecision( + id = null, + requestId = requestId, + outcome = ApprovalOutcome.APPROVED, + state = ApprovalStatus.COMPLETED, + tier = Tier.T2, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = null, + ) + approvalOrchestrator.submitApprovalDecision(requestId, decision) + + runJob.join() val events = eventStore.read(sessionId) - val resumeCount = events.count { it.payload is OrchestrationResumedEvent } - assertEquals(1, resumeCount) + assertEquals(1, events.count { it.payload is OrchestrationResumedEvent }) + assertNotNull(events.find { it.payload is WorkflowCompletedEvent }) } @Test @@ -297,4 +322,5 @@ class SessionOrchestratorIntegrationTest { val state = orchestrationRepository.getState(sessionId) assertEquals(7, state.retryPolicy?.maxAttempts) } + }