fix(server): event capturing and necessary logic for smoke test.

2026-05-17 03:17:12 +04:00
parent 0c1876a549
commit 7d46f46f01
39 changed files with 1097 additions and 67 deletions
@@ -15,9 +15,21 @@ dependencies {
    implementation project(':core:kernel')
    implementation project(':core:inference')
    implementation project(':core:transitions')
+    implementation project(':core:context')
+    implementation project(':core:validation')
+    implementation project(':core:risk')
+    implementation project(':core:artifacts')
+    implementation project(':infrastructure')
+    implementation project(':infrastructure:workflow')
+    implementation project(':infrastructure:persistence')
+    implementation project(':infrastructure:inference')
+    implementation project(':infrastructure:inference:llama_cpp')
+    implementation project(':infrastructure:inference:commons')

    implementation "com.github.ajalt.clikt:clikt:5.0.1"

+    implementation "io.ktor:ktor-client-cio:$ktor_version"
+    implementation "io.ktor:ktor-client-logging:$ktor_version"
    implementation "io.ktor:ktor-server-core:$ktor_version"
    implementation "io.ktor:ktor-server-netty:$ktor_version"
    implementation "io.ktor:ktor-server-websockets:$ktor_version"
@@ -1,5 +1,101 @@
 package com.correx.apps.server

+import com.correx.apps.server.registry.FileSystemWorkflowRegistry
+import com.correx.apps.server.registry.ProviderRegistry
+import com.correx.core.approvals.domain.DefaultApprovalEngine
+import com.correx.core.context.builder.DefaultContextPackBuilder
+import com.correx.core.context.compression.DefaultContextCompressor
+import com.correx.core.inference.DefaultInferenceRouter
+import com.correx.core.inference.InferenceProjector
+import com.correx.core.inference.InferenceRepository
+import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer
+import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
+import com.correx.core.kernel.orchestration.OrchestrationProjector
+import com.correx.core.kernel.orchestration.OrchestrationRepository
+import com.correx.core.kernel.orchestration.OrchestratorEngines
+import com.correx.core.kernel.orchestration.OrchestratorRepositories
+import com.correx.core.kernel.retry.DefaultRetryCoordinator
+import com.correx.core.risk.DefaultRiskAssessor
+import com.correx.core.sessions.DefaultSessionReducer
+import com.correx.core.sessions.DefaultSessionRepository
+import com.correx.core.sessions.SessionProjector
+import com.correx.core.sessions.projections.replay.DefaultEventReplayer
+import com.correx.core.transitions.evaluation.PromptResolver
+import com.correx.core.transitions.resolution.DefaultTransitionResolver
+import com.correx.core.validation.pipeline.ValidationPipeline
+import com.correx.infrastructure.InfrastructureModule
+import com.correx.infrastructure.inference.DefaultProviderRegistry
+import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy
+import io.ktor.server.engine.embeddedServer
+import io.ktor.server.netty.Netty
+
 fun main() {
-    println("correx :: apps/server")
+    val eventStore = InfrastructureModule.createEventStore()
+
+    val llamaProvider = InfrastructureModule.createLlamaCppProvider(
+        modelId = System.getenv("CORREX_MODEL_ID") ?: "default",
+        modelPath = System.getenv("CORREX_MODEL_PATH") ?: "",
+        baseUrl = System.getenv("CORREX_LLAMA_URL") ?: "http://127.0.0.1:10000",
+    )
+    val infraRegistry = InfrastructureModule.createProviderRegistry(listOf(llamaProvider))
+
+    val artifactRepository = InfrastructureModule.createArtifactRepository(eventStore)
+
+    val sessionRepository = DefaultSessionRepository(
+        DefaultEventReplayer(eventStore, SessionProjector(DefaultSessionReducer()))
+    )
+    val inferenceRepository = InferenceRepository(
+        DefaultEventReplayer(eventStore, InferenceProjector())
+    )
+    val orchestrationRepository = OrchestrationRepository(
+        DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer()))
+    )
+
+    val inferenceRouter = DefaultInferenceRouter(infraRegistry, FirstAvailableRoutingStrategy())
+
+    val promptLoader = InfrastructureModule.createPromptLoader()
+    val promptResolver = PromptResolver { path -> promptLoader.load(path) }
+
+    val repositories = OrchestratorRepositories(
+        eventStore = eventStore,
+        inferenceRepository = inferenceRepository,
+        orchestrationRepository = orchestrationRepository,
+        sessionRepository = sessionRepository,
+        artifactRepository = artifactRepository,
+    )
+
+    val engines = OrchestratorEngines(
+        transitionResolver = DefaultTransitionResolver { condition, ctx -> condition.evaluate(ctx) },
+        contextPackBuilder = DefaultContextPackBuilder(DefaultContextCompressor()),
+        inferenceRouter = inferenceRouter,
+        validationPipeline = ValidationPipeline(validators = emptyList()),
+        approvalEngine = DefaultApprovalEngine(),
+        riskAssessor = DefaultRiskAssessor(),
+        promptResolver = promptResolver,
+    )
+
+    val orchestrator = DefaultSessionOrchestrator(
+        repositories = repositories,
+        engines = engines,
+        retryCoordinator = DefaultRetryCoordinator(eventStore),
+    )
+
+    val workflowRegistry = FileSystemWorkflowRegistry(InfrastructureModule.createWorkflowLoader())
+
+    val module = ServerModule(
+        orchestrator = orchestrator,
+        eventStore = eventStore,
+        sessionRepository = sessionRepository,
+        workflowRegistry = workflowRegistry,
+        providerRegistry = infraRegistry.asServerRegistry(),
+    )
+
+    embeddedServer(Netty, port = 8080) {
+        configureServer(module)
+    }.start(wait = true)
+}
+
+private fun DefaultProviderRegistry.asServerRegistry(): ProviderRegistry = object : ProviderRegistry {
+    override fun listAll() = this@asServerRegistry.listAll()
+    override suspend fun healthCheckAll() = this@asServerRegistry.healthCheckAll()
 }
@@ -0,0 +1,39 @@
+package com.correx.apps.server.registry
+
+import com.correx.core.transitions.graph.WorkflowGraph
+import com.correx.infrastructure.workflow.WorkflowLoader
+import java.nio.file.Path
+import kotlin.io.path.exists
+import kotlin.io.path.listDirectoryEntries
+
+class FileSystemWorkflowRegistry(
+    private val loader: WorkflowLoader,
+    private val workflowsDir: Path = Path.of(
+        System.getProperty("user.home"), ".config", "correx", "workflows"
+    ),
+) : WorkflowRegistry {
+
+    private val cache: Map<String, WorkflowGraph> by lazy { loadAll() }
+
+    override fun listAll(): List<WorkflowSummary> =
+        cache.values.map { WorkflowSummary(workflowId = it.id, description = it.id) }
+
+    override fun find(workflowId: String): WorkflowGraph? = cache[workflowId]
+
+    private fun loadAll(): Map<String, WorkflowGraph> {
+        if (!workflowsDir.exists()) return emptyMap()
+        return workflowsDir.listDirectoryEntries("*.toml")
+            .mapNotNull { file ->
+                runCatching { loader.load(file) }
+                    .onFailure { logWarning(file, it) }
+                    .getOrNull()
+            }
+            .associateBy { graph ->
+                graph.id.ifBlank { graph.let { it } .let { file -> file.start.value } }
+            }
+    }
+
+    private fun logWarning(file: Path, ex: Throwable) {
+        System.err.println("[WorkflowRegistry] Skipping $file: ${ex.message}")
+    }
+}
@@ -3,10 +3,16 @@ package com.correx.apps.server.routes
 import com.correx.apps.server.ServerModule
 import com.correx.apps.server.protocol.SessionConfigDto
 import com.correx.apps.server.ws.SessionStreamHandler
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.WorkflowFailedEvent
+import com.correx.core.events.types.EventId
 import com.correx.core.events.types.SessionId
+import com.correx.core.kernel.orchestration.OrchestrationConfig
 import com.correx.core.utils.TypeId
 import io.ktor.http.HttpStatusCode
 import io.ktor.server.application.call
+import io.ktor.server.application.application
 import io.ktor.server.request.receive
 import io.ktor.server.response.respond
 import io.ktor.server.routing.Route
@@ -14,6 +20,8 @@ import io.ktor.server.routing.get
 import io.ktor.server.routing.post
 import io.ktor.server.routing.route
 import io.ktor.server.websocket.webSocket
+import kotlinx.coroutines.launch
+import kotlinx.datetime.Clock
 import kotlinx.serialization.Serializable
 import java.util.UUID

@@ -34,67 +42,95 @@ data class StartSessionResponse(val sessionId: String)

 fun Route.sessionRoutes(module: ServerModule) {
    val streamHandler = SessionStreamHandler(module)
-
    route("/sessions") {
-        get {
-            val sessions = emptyList<SessionSummaryResponse>()
-            call.respond(sessions)
-        }
-
-        post {
-            val body = call.receive<StartSessionRequest>()
-            module.workflowRegistry.find(body.workflowId)
-                ?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: ${body.workflowId}")
-
-            val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
-            call.respond(HttpStatusCode.Created, StartSessionResponse(sessionId.value))
-        }
-
+        get { call.respond(emptyList<SessionSummaryResponse>()) }
+        startSessionRoute(module)
        route("/{id}") {
-            get {
-                val id = call.parameters["id"]
-                    ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id")
-                val sessionId: SessionId = TypeId(id)
-                val session = runCatching { module.sessionRepository.getSession(sessionId) }.getOrNull()
-                    ?: return@get call.respond(HttpStatusCode.NotFound, "Session not found")
-                val response = SessionStateResponse(
-                    sessionId = session.sessionId.value,
-                    status = session.state.status.name,
-                    createdAt = session.state.createdAt?.toString(),
-                )
-                call.respond(response)
-            }
-
-            post("/cancel") {
-                val id = call.parameters["id"]
-                    ?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id")
-                val sessionId: SessionId = TypeId(id)
-                module.orchestrator.cancel(sessionId)
-                call.respond(HttpStatusCode.OK)
-            }
-
-            get("/events") {
-                val id = call.parameters["id"]
-                    ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id")
-                val from = call.request.queryParameters["from"]?.toLongOrNull() ?: 0L
-                val sessionId: SessionId = TypeId(id)
-                val events = module.eventStore.readFrom(sessionId, from)
-                val response = events.map { e ->
-                    EventResponse(
-                        eventId = e.metadata.eventId.value,
-                        sequence = e.sequence,
-                        sessionId = e.metadata.sessionId.value,
-                    )
-                }
-                call.respond(response)
-            }
-
+            getSessionRoute(module)
+            cancelSessionRoute(module)
+            getEventsRoute(module)
            webSocket("/stream") {
                val id = call.parameters["id"] ?: return@webSocket
                val lastEventId = call.request.queryParameters["lastEventId"]?.toLongOrNull()
-                val sessionId: SessionId = TypeId(id)
-                streamHandler.handle(this, sessionId, lastEventId)
+                streamHandler.handle(this, TypeId(id), lastEventId)
            }
        }
    }
 }
+
+private fun Route.startSessionRoute(module: ServerModule) {
+    post {
+        val body = call.receive<StartSessionRequest>()
+        val graph = module.workflowRegistry.find(body.workflowId)
+            ?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: ${body.workflowId}")
+        val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
+        call.application.launch {
+            runCatching { module.orchestrator.run(sessionId, graph, OrchestrationConfig()) }
+                .onFailure { ex ->
+                    System.err.println("[SessionRoutes] run failed for $sessionId: ${ex.message}")
+                    module.eventStore.append(
+                        NewEvent(
+                            metadata = EventMetadata(
+                                eventId = EventId(UUID.randomUUID().toString()),
+                                sessionId = sessionId,
+                                timestamp = Clock.System.now(),
+                                schemaVersion = 1,
+                                causationId = null,
+                                correlationId = null,
+                            ),
+                            payload = WorkflowFailedEvent(
+                                sessionId = sessionId,
+                                stageId = graph.start,
+                                reason = ex.message ?: "unexpected orchestrator failure",
+                                retryExhausted = false,
+                            ),
+                        )
+                    )
+                }
+        }
+        call.respond(HttpStatusCode.Accepted, StartSessionResponse(sessionId.value))
+    }
+}
+
+private fun Route.getSessionRoute(module: ServerModule) {
+    get {
+        val id = call.parameters["id"]
+            ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id")
+        val session = runCatching { module.sessionRepository.getSession(TypeId(id)) }.getOrNull()
+            ?: return@get call.respond(HttpStatusCode.NotFound, "Session not found")
+        call.respond(
+            SessionStateResponse(
+                sessionId = session.sessionId.value,
+                status = session.state.status.name,
+                createdAt = session.state.createdAt?.toString(),
+            )
+        )
+    }
+}
+
+private fun Route.cancelSessionRoute(module: ServerModule) {
+    post("/cancel") {
+        val id = call.parameters["id"]
+            ?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id")
+        module.orchestrator.cancel(TypeId(id))
+        call.respond(HttpStatusCode.OK)
+    }
+}
+
+private fun Route.getEventsRoute(module: ServerModule) {
+    get("/events") {
+        val id = call.parameters["id"]
+            ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id")
+        val from = call.request.queryParameters["from"]?.toLongOrNull() ?: 0L
+        val events = module.eventStore.readFrom(TypeId(id), from)
+        call.respond(
+            events.map { e ->
+                EventResponse(
+                    eventId = e.metadata.eventId.value,
+                    sequence = e.sequence,
+                    sessionId = e.metadata.sessionId.value,
+                )
+            }
+        )
+    }
+}
@@ -3,7 +3,16 @@ package com.correx.apps.server.ws
 import com.correx.apps.server.ServerModule
 import com.correx.apps.server.protocol.ClientMessage
 import com.correx.apps.server.protocol.ProtocolSerializer
+import com.correx.apps.server.protocol.ProviderHealthDto
 import com.correx.apps.server.protocol.ServerMessage
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.WorkflowFailedEvent
+import com.correx.core.events.types.EventId
+import com.correx.core.events.types.SessionId
+import com.correx.core.inference.ProviderHealth
+import com.correx.core.kernel.orchestration.OrchestrationConfig
+import com.correx.core.utils.TypeId
 import io.ktor.server.websocket.DefaultWebSocketServerSession
 import io.ktor.websocket.Frame
 import io.ktor.websocket.readText
@@ -11,12 +20,15 @@ import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import kotlinx.datetime.Clock
+import java.util.*

 private const val HEARTBEAT_INTERVAL_MS = 30_000L

 class GlobalStreamHandler(private val module: ServerModule) {

    suspend fun handle(session: DefaultWebSocketServerSession) {
+        System.err.println("[GlobalStream] client connected")
        sendInitialSnapshot(session)

        val heartbeatJob = session.launch {
@@ -32,26 +44,39 @@ class GlobalStreamHandler(private val module: ServerModule) {
                if (frame is Frame.Text) {
                    val text = frame.readText()
                    runCatching { ProtocolSerializer.decodeClientMessage(text) }
-                        .onSuccess { msg -> handleClientMessage(session, msg) }
+                        .onSuccess { msg ->
+                            System.err.println("[GlobalStream] recv: ${msg::class.simpleName}")
+                            handleClientMessage(session, msg)
+                        }
                        .onFailure {
+                            System.err.println("[GlobalStream] decode error: ${it.message}")
                            val error = ServerMessage.ProtocolError("Unknown message: ${it.message}")
                            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
                        }
                }
            }
        } catch (_: ClosedReceiveChannelException) {
-            // client disconnected
+            System.err.println("[GlobalStream] client disconnected")
        } finally {
            heartbeatJob.cancel()
        }
    }

    private suspend fun sendInitialSnapshot(session: DefaultWebSocketServerSession) {
-        val providerHealth = runCatching { module.providerRegistry.healthCheckAll() }.getOrDefault(emptyMap())
-        providerHealth.forEach { (providerId, _) ->
+        val providerHealth = runCatching { module.providerRegistry.healthCheckAll() }
+            .onFailure { System.err.println("[GlobalStream] healthCheckAll failed: ${it.message}") }
+            .getOrDefault(emptyMap())
+        System.err.println("[GlobalStream] snapshot: ${providerHealth.size} provider(s)")
+        providerHealth.forEach { (providerId, health) ->
+            val status = when (health) {
+                is ProviderHealth.Healthy -> "healthy"
+                is ProviderHealth.Degraded -> "degraded"
+                is ProviderHealth.Unavailable -> "unavailable"
+            }
+            System.err.println("[GlobalStream] provider=${providerId.value} status=$status")
            val msg = ServerMessage.ProviderStatusChanged(
                providerId = providerId.value,
-                status = com.correx.apps.server.protocol.ProviderHealthDto(providerId.value, "unknown", null),
+                status = ProviderHealthDto(providerId.value, status, null),
            )
            session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg)))
        }
@@ -60,8 +85,55 @@ class GlobalStreamHandler(private val module: ServerModule) {
    private suspend fun handleClientMessage(session: DefaultWebSocketServerSession, msg: ClientMessage) {
        when (msg) {
            is ClientMessage.Ping -> Unit
-            else -> {
-                val error = ServerMessage.ProtocolError("Unexpected message type in global stream")
+
+            is ClientMessage.StartSession -> {
+                val graph = module.workflowRegistry.find(msg.workflowId)
+                if (graph == null) {
+                    val error = ServerMessage.ProtocolError("Unknown workflowId: ${msg.workflowId}")
+                    session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
+                    return
+                }
+                val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
+                System.err.println("[GlobalStream] starting session=${sessionId.value} workflow=${msg.workflowId}")
+                session.launch {
+                    runCatching { module.orchestrator.run(sessionId, graph, OrchestrationConfig()) }
+                        .onFailure { ex ->
+                            System.err.println("[GlobalStreamHandler] run failed for $sessionId: ${ex.message}")
+                            module.eventStore.append(
+                                NewEvent(
+                                    metadata = EventMetadata(
+                                        eventId = EventId(UUID.randomUUID().toString()),
+                                        sessionId = sessionId,
+                                        timestamp = Clock.System.now(),
+                                        schemaVersion = 1,
+                                        causationId = null,
+                                        correlationId = null,
+                                    ),
+                                    payload = WorkflowFailedEvent(
+                                        sessionId = sessionId,
+                                        stageId = graph.start,
+                                        reason = ex.message ?: "unexpected orchestrator failure",
+                                        retryExhausted = false,
+                                    ),
+                                ),
+                            )
+                        }
+                }
+                val started = ServerMessage.SessionStarted(sessionId = sessionId, workflowId = msg.workflowId)
+                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started)))
+            }
+
+            is ClientMessage.CancelSession -> {
+                module.orchestrator.cancel(msg.sessionId)
+            }
+
+            is ClientMessage.ResumeSession -> {
+                val error = ServerMessage.ProtocolError("ResumeSession not supported")
+                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
+            }
+
+            is ClientMessage.ApprovalResponse -> {
+                val error = ServerMessage.ProtocolError("ApprovalResponse not supported on global stream")
                session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
            }
        }