debt(pre-router): resolved most of the technical debt that was noted while finishing epic-12 and epic-13.

2026-05-16 14:55:45 +04:00
parent 2207a37549
commit f77efce10b
10 changed files with 107 additions and 43 deletions
@@ -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,
 ) {
@@ -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)
            }
@@ -6,4 +6,5 @@ plugins {

 dependencies {
    implementation "org.jetbrains.kotlinx:kotlinx-datetime"
+    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version"
 }
@@ -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<StoredEvent>
 }
@@ -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)
+}
@@ -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<SessionId, AtomicBoolean> =
        ConcurrentHashMap<SessionId, AtomicBoolean>()

@@ -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)
@@ -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"
@@ -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<SessionId, MutableList<StoredEvent>>()
    private val sequences = ConcurrentHashMap<SessionId, AtomicLong>()
    private val seenEventIds = ConcurrentHashMap.newKeySet<EventId>()
+    private val subscriptions = ConcurrentHashMap<SessionId, MutableSharedFlow<StoredEvent>>()

    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<StoredEvent> =
+        subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) }
+
    private fun doAppend(event: NewEvent, stream: MutableList<StoredEvent>): 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
    }
@@ -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<SessionId, MutableSharedFlow<StoredEvent>>()

    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<NewEvent>): List<StoredEvent> {
@@ -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<StoredEvent> =
@@ -139,6 +146,9 @@ class SqliteEventStore(
            }
        }

+    override fun subscribe(sessionId: SessionId): Flow<StoredEvent> =
+        subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) }
+
    // ---------- helpers ----------

    private fun nextSequence(sessionId: SessionId): Long =
@@ -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<ApprovalGrant>,
-                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)
    }
+
 }