diff --git a/.gitignore b/.gitignore index ba06cb97..6aa91634 100644 --- a/.gitignore +++ b/.gitignore @@ -64,3 +64,4 @@ bin/ /docs/future/ /docs/plans/ /docs/refactor.md +/kls_database.db diff --git a/apps/cli/build.gradle b/apps/cli/build.gradle index 8ac8456d..d8f1fbce 100644 --- a/apps/cli/build.gradle +++ b/apps/cli/build.gradle @@ -1,5 +1,6 @@ plugins { id 'org.jetbrains.kotlin.jvm' + id 'org.jetbrains.kotlin.plugin.serialization' id 'application' } @@ -9,4 +10,10 @@ application { dependencies { implementation "com.github.ajalt.clikt:clikt:5.0.1" + + implementation "io.ktor:ktor-client-core:$ktor_version" + implementation "io.ktor:ktor-client-cio:$ktor_version" + implementation "io.ktor:ktor-client-content-negotiation:$ktor_version" + implementation "io.ktor:ktor-client-websockets:$ktor_version" + implementation "io.ktor:ktor-serialization-kotlinx-json:$ktor_version" } diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/CliConstants.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/CliConstants.kt new file mode 100644 index 00000000..edf66186 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/CliConstants.kt @@ -0,0 +1,3 @@ +package com.correx.apps.cli + +internal const val DEFAULT_PORT = 8080 diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt new file mode 100644 index 00000000..5f073929 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt @@ -0,0 +1,26 @@ +package com.correx.apps.cli + +import com.correx.apps.cli.commands.ApproveCommand +import com.correx.apps.cli.commands.ProviderCommand +import com.correx.apps.cli.commands.RunCommand +import com.correx.apps.cli.commands.SessionCommand +import com.correx.apps.cli.commands.StatusCommand +import com.github.ajalt.clikt.core.CliktCommand +import com.github.ajalt.clikt.core.subcommands +import com.github.ajalt.clikt.parameters.options.flag +import com.github.ajalt.clikt.parameters.options.option + +class CorrexCli : CliktCommand(name = "correx") { + val json by option("--json", help = "Output machine-readable JSON").flag() + val quiet by option("--quiet", help = "Output errors only").flag() + + override fun run() = Unit +} + +fun buildCli(): CorrexCli = CorrexCli().subcommands( + RunCommand(), + SessionCommand(), + ApproveCommand(), + StatusCommand(), + ProviderCommand(), +) diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/Main.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/Main.kt index df654d8c..b6f3de8f 100644 --- a/apps/cli/src/main/kotlin/com/correx/apps/cli/Main.kt +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/Main.kt @@ -1,5 +1,7 @@ package com.correx.apps.cli -fun main() { - println("correx :: apps/cli") +import com.github.ajalt.clikt.core.main + +fun main(args: Array) { + buildCli().main(args) } diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ApproveCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ApproveCommand.kt new file mode 100644 index 00000000..667f53c7 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ApproveCommand.kt @@ -0,0 +1,67 @@ +package com.correx.apps.cli.commands + +import com.correx.apps.cli.CorrexCli +import com.correx.apps.cli.DEFAULT_PORT +import com.github.ajalt.clikt.core.CliktCommand +import com.github.ajalt.clikt.parameters.arguments.argument +import com.github.ajalt.clikt.parameters.options.default +import com.github.ajalt.clikt.parameters.options.option +import com.github.ajalt.clikt.parameters.options.required +import io.ktor.client.HttpClient +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.contentnegotiation.ContentNegotiation +import io.ktor.client.request.post +import io.ktor.client.request.setBody +import io.ktor.http.ContentType +import io.ktor.http.contentType +import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json + +@Serializable +private data class ApprovalRequest( + val decision: String, + val steeringNote: String?, +) + +class ApproveCommand : CliktCommand(name = "approve") { + private val sessionId by argument("SESSION_ID") + private val decision by option("--decision", help = "approve | reject | steer").required() + private val note by option("--note", help = "Steering note") + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + private val approveJson = Json { ignoreUnknownKeys = true } + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = (currentContext.findRoot().command as? CorrexCli)?.json ?: false + + val decisionUpper = decision.uppercase() + if (decisionUpper !in setOf("APPROVE", "REJECT", "STEER")) { + System.err.println("Invalid decision: $decision. Must be approve, reject, or steer.") + return@runBlocking + } + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(approveJson) } + } + + runCatching { + client.post("http://$host:$portInt/sessions/$sessionId/approve") { + contentType(ContentType.Application.Json) + setBody(ApprovalRequest(decision = decisionUpper, steeringNote = note)) + } + if (outputJson) { + println("{\"sessionId\": \"$sessionId\", \"decision\": \"$decisionUpper\"}") + } else { + println("Approval decision '$decisionUpper' sent for session $sessionId.") + } + }.getOrElse { e -> + System.err.println("Error sending approval: ${e.message}") + } + + client.close() + } +} diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ProviderCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ProviderCommand.kt new file mode 100644 index 00000000..a38e3574 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ProviderCommand.kt @@ -0,0 +1,63 @@ +package com.correx.apps.cli.commands + +import com.correx.apps.cli.CorrexCli +import com.correx.apps.cli.DEFAULT_PORT +import com.github.ajalt.clikt.core.CliktCommand +import com.github.ajalt.clikt.core.subcommands +import com.github.ajalt.clikt.parameters.options.default +import com.github.ajalt.clikt.parameters.options.option +import io.ktor.client.HttpClient +import io.ktor.client.call.body +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.contentnegotiation.ContentNegotiation +import io.ktor.client.request.get +import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.jsonPrimitive + +class ProviderCommand : CliktCommand(name = "provider") { + override fun run() = Unit + + init { + subcommands(ProviderListCommand()) + } +} + +class ProviderListCommand : CliktCommand(name = "list") { + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + private val providerJson = Json { ignoreUnknownKeys = true } + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = (currentContext.findRoot().command as? CorrexCli)?.json ?: false + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(providerJson) } + } + + runCatching { + val resp = client.get("http://$host:$portInt/providers") + val body = resp.body() + if (outputJson) { + println(body.toString()) + } else { + body.forEach { el -> + val obj = el as? JsonObject ?: return@forEach + val providerId = obj["providerId"]?.jsonPrimitive?.content ?: "" + val status = obj["status"]?.jsonPrimitive?.content ?: "" + val load = obj["loadPercent"]?.jsonPrimitive?.content ?: "?" + println("$providerId status=$status load=${load}%") + } + } + }.getOrElse { e -> + System.err.println("Error fetching providers: ${e.message}") + } + + client.close() + } +} diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/RunCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/RunCommand.kt new file mode 100644 index 00000000..2b6f4235 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/RunCommand.kt @@ -0,0 +1,256 @@ +package com.correx.apps.cli.commands + +import com.correx.apps.cli.CorrexCli +import com.correx.apps.cli.DEFAULT_PORT +import com.correx.apps.cli.ws.CliWsClient +import com.github.ajalt.clikt.core.CliktCommand +import com.github.ajalt.clikt.parameters.options.default +import com.github.ajalt.clikt.parameters.options.flag +import com.github.ajalt.clikt.parameters.options.option +import com.github.ajalt.clikt.parameters.options.required +import io.ktor.client.HttpClient +import io.ktor.client.call.body +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.contentnegotiation.ContentNegotiation +import io.ktor.client.plugins.websocket.WebSockets +import io.ktor.client.request.post +import io.ktor.client.request.setBody +import io.ktor.http.ContentType +import io.ktor.http.contentType +import io.ktor.serialization.kotlinx.json.json +import io.ktor.websocket.WebSocketSession +import io.ktor.websocket.send +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.Serializable +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive + +@Serializable +private data class StartSessionRequest(val workflowId: String, val sessionId: String?) + +@Serializable +private data class StartSessionResponse(val sessionId: String) + +@Serializable +private data class ApprovalResponsePayload( + val type: String, + val requestId: String, + val decision: String, + val steeringNote: String?, +) + +private data class RunContext( + val sid: String, + val outputJson: Boolean, + val autoApprove: Boolean, + val isTty: Boolean, +) + +private data class ApprovalContext( + val session: WebSocketSession, + val requestId: String, + val toolName: String, + val preview: String?, +) + +class RunCommand : CliktCommand(name = "run") { + private val workflow by option("--workflow", help = "Path to workflow definition").required() + private val sessionId by option("--session", help = "Existing session ID to resume") + private val autoApprove by option("--auto-approve", help = "Auto-approve all approval requests").flag() + private val host by option("--host", help = "Server host").default("localhost") + private val port by option("--port", help = "Server port").default("$DEFAULT_PORT") + + private val serializer = Json { ignoreUnknownKeys = true; classDiscriminator = "type" } + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = (currentContext.findRoot().command as? CorrexCli)?.json ?: false + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(serializer) } + install(WebSockets) + } + + val sid = startSession(client, host, portInt, outputJson) ?: run { + client.close() + return@runBlocking + } + + printLine("session_started", mapOf("sessionId" to sid, "workflow" to workflow), outputJson) + + val ctx = RunContext(sid, outputJson, autoApprove, System.console() != null) + var exitCode = 0 + val wsClient = CliWsClient(client) + + wsClient.sessionStream(host, portInt, sid) { session, raw -> + val msgJson = runCatching { serializer.parseToJsonElement(raw).jsonObject }.getOrNull() + val keepGoing = handleMessage(session, msgJson, ctx) { exitCode = it } + keepGoing + } + + client.close() + if (exitCode != 0) { + throw SystemExitException(exitCode) + } + } + + private suspend fun startSession( + client: HttpClient, + host: String, + portInt: Int, + outputJson: Boolean, + ): String? = runCatching { + val resp = client.post("http://$host:$portInt/sessions") { + contentType(ContentType.Application.Json) + setBody(StartSessionRequest(workflowId = workflow, sessionId = sessionId)) + } + resp.body().sessionId + }.getOrElse { e -> + printLine("error", mapOf("message" to "Failed to start session: ${e.message}"), outputJson) + null + } + + @Suppress("CyclomaticComplexMethod", "LongMethod") + private suspend fun handleMessage( + session: WebSocketSession, + msgJson: JsonObject?, + ctx: RunContext, + setExitCode: (Int) -> Unit, + ): Boolean { + val type = msgJson?.get("type")?.jsonPrimitive?.content ?: "unknown" + fun field(key: String): String? = msgJson?.get(key)?.jsonPrimitive?.content + + return when (type) { + "SessionCompleted" -> { + printLine("session_completed", mapOf("sessionId" to ctx.sid), ctx.outputJson) + setExitCode(0) + false + } + "SessionFailed" -> { + val reason = field("reason") ?: "unknown" + printLine("session_failed", mapOf("sessionId" to ctx.sid, "reason" to reason), ctx.outputJson) + setExitCode(1) + false + } + "StageStarted" -> { + printLine("stage_started", mapOf("stageId" to (field("stageId") ?: "")), ctx.outputJson) + true + } + "StageCompleted" -> { + printLine("stage_completed", mapOf("stageId" to (field("stageId") ?: "")), ctx.outputJson) + true + } + "StageFailed" -> { + printLine( + "stage_failed", + mapOf("stageId" to (field("stageId") ?: ""), "reason" to (field("reason") ?: "")), + ctx.outputJson, + ) + true + } + "ToolStarted" -> { + printLine("tool_started", mapOf("tool" to (field("toolName") ?: "")), ctx.outputJson) + true + } + "ToolCompleted" -> { + printLine("tool_completed", mapOf("tool" to (field("toolName") ?: "")), ctx.outputJson) + true + } + "ToolFailed" -> { + printLine( + "tool_failed", + mapOf("tool" to (field("toolName") ?: ""), "reason" to (field("reason") ?: "")), + ctx.outputJson, + ) + true + } + "ToolRejected" -> { + printLine( + "tool_rejected", + mapOf("tool" to (field("toolName") ?: ""), "reason" to (field("reason") ?: "")), + ctx.outputJson, + ) + true + } + "ApprovalRequired" -> { + val approvalCtx = ApprovalContext( + session = session, + requestId = field("requestId") ?: "", + toolName = field("toolName") ?: "", + preview = field("preview"), + ) + handleApproval(approvalCtx, ctx, setExitCode) + } + else -> true + } + } + + private suspend fun handleApproval( + approvalCtx: ApprovalContext, + ctx: RunContext, + setExitCode: (Int) -> Unit, + ): Boolean { + val decision = resolveDecision(ctx, approvalCtx.toolName, approvalCtx.preview) + + val steeringNote = if (decision == "STEER" && ctx.isTty) { + print("Steering note: ") + readLine() + } else { + null + } + + val response = ApprovalResponsePayload( + type = "ApprovalResponse", + requestId = approvalCtx.requestId, + decision = decision, + steeringNote = steeringNote, + ) + approvalCtx.session.send(serializer.encodeToString(response)) + + return if (decision == "REJECT" && !ctx.autoApprove && !ctx.isTty) { + setExitCode(2) + false + } else { + true + } + } + + private fun resolveDecision(ctx: RunContext, toolName: String, preview: String?): String = when { + ctx.autoApprove -> "APPROVE" + !ctx.isTty -> { + System.err.println( + "WARNING: approval required but no TTY and --auto-approve not set — denying" + ) + "REJECT" + } + else -> promptApproval(toolName, preview) + } + + private fun promptApproval(toolName: String, preview: String?): String { + println("\nApproval required for tool: $toolName") + preview?.let { println("Preview: $it") } + print("[a]pprove / [r]eject / [s]teer: ") + return when (readLine()?.trim()?.lowercase()) { + "a", "approve" -> "APPROVE" + "s", "steer" -> "STEER" + else -> "REJECT" + } + } + + private fun printLine(event: String, data: Map, outputJson: Boolean) { + if (outputJson) { + val fields = data.entries.joinToString(", ") { (k, v) -> + "\"$k\": ${if (v == null) "null" else "\"$v\""}" + } + println("{\"event\": \"$event\", $fields}") + } else { + val details = data.entries.joinToString(" ") { (k, v) -> "$k=${v ?: "null"}" } + println("[$event] $details") + } + } +} + +class SystemExitException(val code: Int) : Exception("exit $code") diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/SessionCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/SessionCommand.kt new file mode 100644 index 00000000..befad296 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/SessionCommand.kt @@ -0,0 +1,164 @@ +package com.correx.apps.cli.commands + +import com.correx.apps.cli.CorrexCli +import com.correx.apps.cli.DEFAULT_PORT +import com.github.ajalt.clikt.core.CliktCommand +import com.github.ajalt.clikt.core.subcommands +import com.github.ajalt.clikt.parameters.arguments.argument +import com.github.ajalt.clikt.parameters.options.default +import com.github.ajalt.clikt.parameters.options.option +import io.ktor.client.HttpClient +import io.ktor.client.call.body +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.contentnegotiation.ContentNegotiation +import io.ktor.client.request.get +import io.ktor.client.request.post +import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.jsonPrimitive + +class SessionCommand : CliktCommand(name = "session") { + override fun run() = Unit + + init { + subcommands( + SessionListCommand(), + SessionResumeCommand(), + SessionCancelCommand(), + SessionEventsCommand(), + ) + } +} + +private fun parentJsonFlag(command: CliktCommand): Boolean = + (command.currentContext.findRoot().command as? CorrexCli)?.json ?: false + +private val sessionJson = Json { ignoreUnknownKeys = true } + +class SessionListCommand : CliktCommand(name = "list") { + private val limit by option("--limit", help = "Max sessions to return").default("20") + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = parentJsonFlag(this@SessionListCommand) + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(sessionJson) } + } + + runCatching { + val resp = client.get("http://$host:$portInt/sessions?limit=$limit") + val body = resp.body() + if (outputJson) { + println(body.toString()) + } else { + body.forEach { el -> + val obj = el as? JsonObject ?: return@forEach + val sid = obj["sessionId"]?.jsonPrimitive?.content ?: "" + val wf = obj["workflowId"]?.jsonPrimitive?.content ?: "" + val status = obj["status"]?.jsonPrimitive?.content ?: "" + println("$sid workflow=$wf status=$status") + } + } + }.getOrElse { e -> + System.err.println("Error listing sessions: ${e.message}") + } + + client.close() + } +} + +class SessionResumeCommand : CliktCommand(name = "resume") { + private val id by argument("SESSION_ID") + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = parentJsonFlag(this@SessionResumeCommand) + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(sessionJson) } + } + + runCatching { + client.post("http://$host:$portInt/sessions/$id/resume") + if (outputJson) { + println("{\"sessionId\": \"$id\", \"action\": \"resumed\"}") + } else { + println("Session $id resumed.") + } + }.getOrElse { e -> + System.err.println("Error resuming session: ${e.message}") + } + + client.close() + } +} + +class SessionCancelCommand : CliktCommand(name = "cancel") { + private val id by argument("SESSION_ID") + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = parentJsonFlag(this@SessionCancelCommand) + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(sessionJson) } + } + + runCatching { + client.post("http://$host:$portInt/sessions/$id/cancel") + if (outputJson) { + println("{\"sessionId\": \"$id\", \"action\": \"cancelled\"}") + } else { + println("Session $id cancelled.") + } + }.getOrElse { e -> + System.err.println("Error cancelling session: ${e.message}") + } + + client.close() + } +} + +class SessionEventsCommand : CliktCommand(name = "events") { + private val id by argument("SESSION_ID") + private val from by option("--from", help = "Start from event ID") + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = parentJsonFlag(this@SessionEventsCommand) + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(sessionJson) } + } + + val query = from?.let { "?from=$it" } ?: "" + + runCatching { + val resp = client.get("http://$host:$portInt/sessions/$id/events$query") + val body = resp.body() + if (outputJson) { + println(body.toString()) + } else { + body.forEachIndexed { idx, el -> + println("[$idx] $el") + } + } + }.getOrElse { e -> + System.err.println("Error fetching events: ${e.message}") + } + + client.close() + } +} diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatusCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatusCommand.kt new file mode 100644 index 00000000..fffe0a4b --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatusCommand.kt @@ -0,0 +1,49 @@ +package com.correx.apps.cli.commands + +import com.correx.apps.cli.CorrexCli +import com.correx.apps.cli.DEFAULT_PORT +import com.github.ajalt.clikt.core.CliktCommand +import com.github.ajalt.clikt.parameters.options.default +import com.github.ajalt.clikt.parameters.options.option +import io.ktor.client.HttpClient +import io.ktor.client.call.body +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.contentnegotiation.ContentNegotiation +import io.ktor.client.request.get +import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.jsonPrimitive + +class StatusCommand : CliktCommand(name = "status") { + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + private val statusJson = Json { ignoreUnknownKeys = true } + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = (currentContext.findRoot().command as? CorrexCli)?.json ?: false + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(statusJson) } + } + + runCatching { + val resp = client.get("http://$host:$portInt/status") + val body = resp.body() + if (outputJson) { + println(body.toString()) + } else { + body.entries.forEach { (k, v) -> + println("$k: ${v.jsonPrimitive.content}") + } + } + }.getOrElse { e -> + System.err.println("Error fetching status: ${e.message}") + } + + client.close() + } +} diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/ws/CliWsClient.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/ws/CliWsClient.kt new file mode 100644 index 00000000..add9b0b3 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/ws/CliWsClient.kt @@ -0,0 +1,25 @@ +package com.correx.apps.cli.ws + +import io.ktor.client.HttpClient +import io.ktor.client.plugins.websocket.webSocket +import io.ktor.websocket.Frame +import io.ktor.websocket.WebSocketSession +import io.ktor.websocket.readText + +class CliWsClient(private val httpClient: HttpClient) { + suspend fun sessionStream( + host: String, + port: Int, + sessionId: String, + onMessage: suspend (WebSocketSession, String) -> Boolean, + ) { + httpClient.webSocket(host = host, port = port, path = "/sessions/$sessionId/stream") { + for (frame in incoming) { + if (frame is Frame.Text) { + val keepGoing = onMessage(this, frame.readText()) + if (!keepGoing) break + } + } + } + } +} diff --git a/apps/server/build.gradle b/apps/server/build.gradle index a14ee350..26b23e1d 100644 --- a/apps/server/build.gradle +++ b/apps/server/build.gradle @@ -1,5 +1,6 @@ plugins { id 'org.jetbrains.kotlin.jvm' + id 'org.jetbrains.kotlin.plugin.serialization' id 'application' } @@ -8,5 +9,19 @@ application { } dependencies { + implementation project(':core:events') + implementation project(':core:approvals') + implementation project(':core:sessions') + implementation project(':core:kernel') + implementation project(':core:inference') + implementation project(':core:transitions') + implementation "com.github.ajalt.clikt:clikt:5.0.1" + + implementation "io.ktor:ktor-server-core:$ktor_version" + implementation "io.ktor:ktor-server-netty:$ktor_version" + implementation "io.ktor:ktor-server-websockets:$ktor_version" + implementation "io.ktor:ktor-server-content-negotiation:$ktor_version" + implementation "io.ktor:ktor-serialization-kotlinx-json:$ktor_version" + implementation "io.ktor:ktor-server-status-pages:$ktor_version" } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Application.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Application.kt new file mode 100644 index 00000000..db9b08bd --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Application.kt @@ -0,0 +1,49 @@ +package com.correx.apps.server + +import com.correx.apps.server.routes.providerRoutes +import com.correx.apps.server.routes.sessionRoutes +import com.correx.apps.server.routes.workflowRoutes +import com.correx.apps.server.ws.GlobalStreamHandler +import io.ktor.http.HttpStatusCode +import io.ktor.serialization.kotlinx.json.json +import io.ktor.server.application.Application +import io.ktor.server.application.install +import io.ktor.server.plugins.contentnegotiation.ContentNegotiation +import io.ktor.server.plugins.statuspages.StatusPages +import io.ktor.server.response.respond +import io.ktor.server.routing.get +import io.ktor.server.routing.routing +import io.ktor.server.websocket.WebSockets +import io.ktor.server.websocket.webSocket +import kotlinx.serialization.json.Json + +fun Application.configureServer(module: ServerModule) { + install(WebSockets) + + install(ContentNegotiation) { + json(Json { ignoreUnknownKeys = true }) + } + + install(StatusPages) { + exception { call, cause -> + call.respond(HttpStatusCode.InternalServerError, mapOf("error" to (cause.message ?: "Internal error"))) + } + } + + val globalStreamHandler = GlobalStreamHandler(module) + + routing { + get("/health") { + val providerHealth = runCatching { module.providerRegistry.healthCheckAll() }.getOrDefault(emptyMap()) + call.respond(mapOf("status" to "ok", "providers" to providerHealth.size.toString())) + } + + webSocket("/stream") { + globalStreamHandler.handle(this) + } + + sessionRoutes(module) + workflowRoutes(module) + providerRoutes(module) + } +} 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 new file mode 100644 index 00000000..9df81668 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -0,0 +1,15 @@ +package com.correx.apps.server + +import com.correx.apps.server.registry.ProviderRegistry +import com.correx.apps.server.registry.WorkflowRegistry +import com.correx.core.events.stores.EventStore +import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.sessions.DefaultSessionRepository + +data class ServerModule( + val orchestrator: DefaultSessionOrchestrator, + val eventStore: EventStore, + val sessionRepository: DefaultSessionRepository, + val workflowRegistry: WorkflowRegistry, + val providerRegistry: ProviderRegistry, +) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt new file mode 100644 index 00000000..bc161ec6 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt @@ -0,0 +1,5 @@ +package com.correx.apps.server.approval + +data class ApprovalConfig( + val timeoutMs: Long = 300_000L, +) 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 new file mode 100644 index 00000000..fc2ff020 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt @@ -0,0 +1,137 @@ +package com.correx.apps.server.approval + +import com.correx.apps.server.protocol.ApprovalDecision +import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ProtocolSerializer +import com.correx.apps.server.protocol.RiskSummaryDto +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.Tier +import com.correx.core.approvals.UserSteering +import com.correx.core.approvals.model.ApprovalContext +import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision +import com.correx.core.approvals.model.ApprovalScopeIdentity +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.sessions.ApprovalMode +import io.ktor.server.websocket.DefaultWebSocketServerSession +import io.ktor.websocket.Frame +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.datetime.Clock +import java.util.concurrent.ConcurrentHashMap + +class ApprovalCoordinator( + private val orchestrator: DefaultSessionOrchestrator, + private val config: ApprovalConfig, + private val scope: CoroutineScope, +) { + private val sessionClients: ConcurrentHashMap> = + ConcurrentHashMap() + private val resolved: ConcurrentHashMap = ConcurrentHashMap() + private val timeoutJobs: ConcurrentHashMap = ConcurrentHashMap() + + fun registerClient(sessionId: SessionId, session: DefaultWebSocketServerSession) { + sessionClients.getOrPut(sessionId) { ConcurrentHashMap.newKeySet() }.add(session) + } + + fun unregisterClient(sessionId: SessionId, session: DefaultWebSocketServerSession) { + sessionClients[sessionId]?.remove(session) + } + + suspend fun onApprovalRequested(event: ApprovalRequestedEvent) { + val msg = ServerMessage.ApprovalRequired( + sessionId = event.sessionId, + requestId = event.requestId, + tier = event.tier, + riskSummary = RiskSummaryDto( + level = event.tier.name, + factors = emptyList(), + recommendedAction = "Review and approve or reject", + ), + toolName = null, + preview = null, + ) + broadcast(event.sessionId, msg) + scheduleTimeout(event.requestId, event.sessionId, event.stageId, event.tier) + } + + suspend fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? { + if (resolved.putIfAbsent(msg.requestId, true) != null) { + return ServerMessage.ProtocolError("Approval request ${msg.requestId.value} already resolved") + } + timeoutJobs.remove(msg.requestId)?.cancel() + val domain = msg.toDomainDecision(sessionId, null, Tier.T2) + return runCatching { orchestrator.submitApprovalDecision(msg.requestId, domain) } + .fold(onSuccess = { null }, onFailure = { ServerMessage.ProtocolError(it.message ?: "Unknown error") }) + } + + private suspend fun broadcast(sessionId: SessionId, msg: ServerMessage) { + val encoded = ProtocolSerializer.encodeServerMessage(msg) + sessionClients[sessionId]?.forEach { client -> + runCatching { client.send(Frame.Text(encoded)) } + } + } + + private fun scheduleTimeout( + requestId: ApprovalRequestId, + sessionId: SessionId, + stageId: StageId?, + tier: Tier, + ) { + val job = scope.launch { + delay(config.timeoutMs) + if (resolved.putIfAbsent(requestId, true) != null) return@launch + val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = stageId, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val decision = DomainApprovalDecision( + id = null, + requestId = requestId, + outcome = ApprovalOutcome.REJECTED, + state = ApprovalStatus.COMPLETED, + tier = tier, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = "Approval timed out after ${config.timeoutMs}ms", + ) + runCatching { orchestrator.submitApprovalDecision(requestId, decision) } + } + timeoutJobs[requestId] = job + } + + private fun ClientMessage.ApprovalResponse.toDomainDecision( + sessionId: SessionId, + stageId: StageId?, + tier: Tier, + ): DomainApprovalDecision { + val outcome = when (decision) { + ApprovalDecision.APPROVE -> ApprovalOutcome.APPROVED + ApprovalDecision.REJECT -> ApprovalOutcome.REJECTED + ApprovalDecision.STEER -> ApprovalOutcome.APPROVED + } + val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = stageId, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val steering = if (decision == ApprovalDecision.STEER) { + steeringNote?.let { UserSteering(text = it, sessionId = sessionId, timestamp = Clock.System.now()) } + } else { + null + } + return DomainApprovalDecision( + id = null, + requestId = requestId, + outcome = outcome, + state = ApprovalStatus.COMPLETED, + tier = tier, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = steeringNote, + userSteering = steering, + ) + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt new file mode 100644 index 00000000..94e70e80 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt @@ -0,0 +1,34 @@ +package com.correx.apps.server.protocol + +import com.correx.core.events.types.ApprovalRequestId +import com.correx.core.events.types.SessionId +import kotlinx.serialization.Serializable + +@Serializable +enum class ApprovalDecision { + APPROVE, + REJECT, + STEER, +} + +@Serializable +sealed class ClientMessage { + @Serializable + data class StartSession(val workflowId: String, val config: SessionConfigDto?) : ClientMessage() + + @Serializable + data class ResumeSession(val sessionId: SessionId) : ClientMessage() + + @Serializable + data class CancelSession(val sessionId: SessionId) : ClientMessage() + + @Serializable + data class ApprovalResponse( + val requestId: ApprovalRequestId, + val decision: ApprovalDecision, + val steeringNote: String?, + ) : ClientMessage() + + @Serializable + data class Ping(val timestamp: Long) : ClientMessage() +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt new file mode 100644 index 00000000..d53000d1 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt @@ -0,0 +1,29 @@ +package com.correx.apps.server.protocol + +import kotlinx.serialization.Serializable + +@Serializable +data class RiskSummaryDto( + val level: String, + val factors: List, + val recommendedAction: String, +) + +@Serializable +data class ProviderHealthDto( + val providerId: String, + val status: String, + val loadPercent: Int?, +) + +@Serializable +data class SessionConfigDto( + val timeoutMs: Long?, + val retryPolicy: String?, +) + +@Serializable +enum class PauseReason { + APPROVAL_PENDING, + USER_REQUESTED, +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ProtocolSerializer.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ProtocolSerializer.kt new file mode 100644 index 00000000..c033d12b --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ProtocolSerializer.kt @@ -0,0 +1,24 @@ +package com.correx.apps.server.protocol + +import kotlinx.serialization.json.Json +import kotlinx.serialization.encodeToString +import kotlinx.serialization.decodeFromString + +class ProtocolException(message: String, cause: Throwable? = null) : RuntimeException(message, cause) + +object ProtocolSerializer { + private val json = Json { + classDiscriminator = "type" + ignoreUnknownKeys = true + } + + fun encodeServerMessage(msg: ServerMessage): String = json.encodeToString(msg) + + fun decodeClientMessage(json: String): ClientMessage { + return runCatching { + this.json.decodeFromString(json) + }.getOrElse { cause -> + throw ProtocolException("Failed to decode client message", cause) + } + } +} 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 new file mode 100644 index 00000000..13f1ff29 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -0,0 +1,76 @@ +package com.correx.apps.server.protocol + +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.Serializable + +@Serializable +sealed class ServerMessage { + @Serializable + data class SessionStarted(val sessionId: SessionId, val workflowId: String) : ServerMessage() + + @Serializable + data class SessionPaused(val sessionId: SessionId, val reason: PauseReason) : ServerMessage() + + @Serializable + data class SessionCompleted(val sessionId: SessionId) : ServerMessage() + + @Serializable + data class SessionFailed(val sessionId: SessionId, val reason: String) : ServerMessage() + + @Serializable + data class StageStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + + @Serializable + data class StageCompleted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + + @Serializable + data class StageFailed(val sessionId: SessionId, val stageId: StageId, val reason: String) : + ServerMessage() + + @Serializable + data class InferenceStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + + @Serializable + data class InferenceCompleted(val sessionId: SessionId, val stageId: StageId, val outputSummary: String) : + ServerMessage() + + @Serializable + data class InferenceTimedOut(val sessionId: SessionId, val stageId: StageId, val elapsedMs: Long) : + ServerMessage() + + @Serializable + data class ToolStarted(val sessionId: SessionId, val toolName: String, val tier: Tier) : + ServerMessage() + + @Serializable + data class ToolCompleted(val sessionId: SessionId, val toolName: String, val outputSummary: String) : + ServerMessage() + + @Serializable + data class ToolFailed(val sessionId: SessionId, val toolName: String, val reason: String) : + ServerMessage() + + @Serializable + data class ToolRejected(val sessionId: SessionId, val toolName: String, val reason: String) : + ServerMessage() + + @Serializable + data class ApprovalRequired( + val sessionId: SessionId, + val requestId: ApprovalRequestId, + val tier: Tier, + val riskSummary: RiskSummaryDto, + val toolName: String?, + val preview: String?, + ) : ServerMessage() + + @Serializable + data class ProviderStatusChanged(val providerId: String, val status: ProviderHealthDto) : + ServerMessage() + + @Serializable + data class ProtocolError(val message: String) : ServerMessage() +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/registry/ProviderRegistry.kt b/apps/server/src/main/kotlin/com/correx/apps/server/registry/ProviderRegistry.kt new file mode 100644 index 00000000..71d904b9 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/registry/ProviderRegistry.kt @@ -0,0 +1,10 @@ +package com.correx.apps.server.registry + +import com.correx.core.inference.InferenceProvider +import com.correx.core.inference.ProviderHealth +import com.correx.core.events.types.ProviderId + +interface ProviderRegistry { + fun listAll(): List + suspend fun healthCheckAll(): Map +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/registry/WorkflowRegistry.kt b/apps/server/src/main/kotlin/com/correx/apps/server/registry/WorkflowRegistry.kt new file mode 100644 index 00000000..d1a0b972 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/registry/WorkflowRegistry.kt @@ -0,0 +1,13 @@ +package com.correx.apps.server.registry + +import com.correx.core.transitions.graph.WorkflowGraph + +interface WorkflowRegistry { + fun listAll(): List + fun find(workflowId: String): WorkflowGraph? +} + +data class WorkflowSummary( + val workflowId: String, + val description: String, +) 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 new file mode 100644 index 00000000..97edd8ac --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/ApprovalRoutes.kt @@ -0,0 +1,52 @@ +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.core.events.types.ApprovalRequestId +import com.correx.core.events.types.SessionId +import com.correx.core.utils.TypeId +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.call +import io.ktor.server.request.receive +import io.ktor.server.response.respond +import io.ktor.server.routing.Route +import io.ktor.server.routing.post +import io.ktor.server.routing.route +import kotlinx.serialization.Serializable + +@Serializable +data class ApprovalRequest( + val requestId: String, + val decision: String, + val note: String? = null, +) + +fun Route.approvalRoutes(coordinator: ApprovalCoordinator) { + route("/sessions/{id}/approve") { + post { + val id = call.parameters["id"] + ?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id") + val sessionId: SessionId = TypeId(id) + val body = call.receive() + val requestId: ApprovalRequestId = TypeId(body.requestId) + val decision = when (body.decision.lowercase()) { + "approve" -> ApprovalDecision.APPROVE + "reject" -> ApprovalDecision.REJECT + "steer" -> ApprovalDecision.STEER + else -> return@post call.respond(HttpStatusCode.BadRequest, "Unknown decision: ${body.decision}") + } + val msg = ClientMessage.ApprovalResponse( + requestId = requestId, + decision = decision, + steeringNote = body.note, + ) + 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)) + } else { + call.respond(HttpStatusCode.OK) + } + } + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/ProviderRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/ProviderRoutes.kt new file mode 100644 index 00000000..21c1b841 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/ProviderRoutes.kt @@ -0,0 +1,31 @@ +package com.correx.apps.server.routes + +import com.correx.apps.server.ServerModule +import com.correx.apps.server.protocol.ProviderHealthDto +import com.correx.core.inference.ProviderHealth +import io.ktor.server.application.call +import io.ktor.server.response.respond +import io.ktor.server.routing.Route +import io.ktor.server.routing.get +import kotlinx.serialization.Serializable + +fun Route.providerRoutes(module: ServerModule) { + get("/providers") { + val providers = module.providerRegistry.listAll() + val healthMap = runCatching { module.providerRegistry.healthCheckAll() }.getOrDefault(emptyMap()) + val response = providers.map { provider -> + val health = healthMap[provider.id] + val dto = when (health) { + is ProviderHealth.Healthy -> ProviderHealthDto(provider.id.value, "healthy", null) + is ProviderHealth.Degraded -> ProviderHealthDto(provider.id.value, "degraded", null) + is ProviderHealth.Unavailable -> ProviderHealthDto(provider.id.value, "unavailable", null) + null -> ProviderHealthDto(provider.id.value, "unknown", null) + } + ProviderStatusEntry(providerId = provider.id.value, name = provider.name, health = dto) + } + call.respond(response) + } +} + +@Serializable +data class ProviderStatusEntry(val providerId: String, val name: String, val health: ProviderHealthDto) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt new file mode 100644 index 00000000..ac01283c --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt @@ -0,0 +1,100 @@ +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.types.SessionId +import com.correx.core.utils.TypeId +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.call +import io.ktor.server.request.receive +import io.ktor.server.response.respond +import io.ktor.server.routing.Route +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.serialization.Serializable +import java.util.UUID + +@Serializable +data class StartSessionRequest(val workflowId: String, val config: SessionConfigDto? = null) + +@Serializable +data class SessionSummaryResponse(val sessionId: String, val status: String) + +@Serializable +data class SessionStateResponse(val sessionId: String, val status: String, val createdAt: String?) + +@Serializable +data class EventResponse(val eventId: String, val sequence: Long, val sessionId: String) + +@Serializable +data class StartSessionResponse(val sessionId: String) + +fun Route.sessionRoutes(module: ServerModule) { + val streamHandler = SessionStreamHandler(module) + + route("/sessions") { + get { + val sessions = emptyList() + call.respond(sessions) + } + + post { + val body = call.receive() + 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)) + } + + 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) + } + + 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) + } + } + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/WorkflowRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/WorkflowRoutes.kt new file mode 100644 index 00000000..37e9ca1c --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/WorkflowRoutes.kt @@ -0,0 +1,15 @@ +package com.correx.apps.server.routes + +import com.correx.apps.server.ServerModule +import com.correx.apps.server.registry.WorkflowSummary +import io.ktor.server.application.call +import io.ktor.server.response.respond +import io.ktor.server.routing.Route +import io.ktor.server.routing.get + +fun Route.workflowRoutes(module: ServerModule) { + get("/workflows") { + val workflows: List = module.workflowRegistry.listAll() + call.respond(workflows) + } +} 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 new file mode 100644 index 00000000..d2393b32 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -0,0 +1,69 @@ +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.ServerMessage +import io.ktor.server.websocket.DefaultWebSocketServerSession +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.channels.ClosedReceiveChannelException +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch + +private const val HEARTBEAT_INTERVAL_MS = 30_000L + +class GlobalStreamHandler(private val module: ServerModule) { + + suspend fun handle(session: DefaultWebSocketServerSession) { + sendInitialSnapshot(session) + + val heartbeatJob = session.launch { + while (isActive) { + delay(HEARTBEAT_INTERVAL_MS) + val ping = ServerMessage.ProtocolError("ping") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(ping))) + } + } + + try { + for (frame in session.incoming) { + if (frame is Frame.Text) { + val text = frame.readText() + runCatching { ProtocolSerializer.decodeClientMessage(text) } + .onSuccess { msg -> handleClientMessage(session, msg) } + .onFailure { + val error = ServerMessage.ProtocolError("Unknown message: ${it.message}") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) + } + } + } + } catch (_: ClosedReceiveChannelException) { + // client disconnected + } finally { + heartbeatJob.cancel() + } + } + + private suspend fun sendInitialSnapshot(session: DefaultWebSocketServerSession) { + val providerHealth = runCatching { module.providerRegistry.healthCheckAll() }.getOrDefault(emptyMap()) + providerHealth.forEach { (providerId, _) -> + val msg = ServerMessage.ProviderStatusChanged( + providerId = providerId.value, + status = com.correx.apps.server.protocol.ProviderHealthDto(providerId.value, "unknown", null), + ) + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg))) + } + } + + 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") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) + } + } + } +} 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 new file mode 100644 index 00000000..e225803d --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt @@ -0,0 +1,116 @@ +package com.correx.apps.server.ws + +import com.correx.apps.server.ServerModule +import com.correx.apps.server.protocol.ApprovalDecision +import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ProtocolSerializer +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.Tier +import com.correx.core.approvals.model.ApprovalContext +import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision +import com.correx.core.approvals.model.ApprovalScopeIdentity +import com.correx.core.approvals.UserSteering +import com.correx.core.events.types.SessionId +import com.correx.core.sessions.ApprovalMode +import com.correx.core.utils.TypeId +import io.ktor.server.websocket.DefaultWebSocketServerSession +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.channels.ClosedReceiveChannelException +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.datetime.Clock + +private const val HEARTBEAT_INTERVAL_MS = 30_000L + +class SessionStreamHandler(private val module: ServerModule) { + + suspend fun handle(session: DefaultWebSocketServerSession, sessionId: SessionId, lastEventId: Long?) { + val replayEvents = if (lastEventId != null) { + module.eventStore.readFrom(sessionId, lastEventId) + } else { + emptyList() + } + + runCatching { module.sessionRepository.getSession(sessionId) }.onSuccess { + val snapshot = ServerMessage.SessionStarted(sessionId, sessionId.value) + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(snapshot))) + } + + for (event in replayEvents) { + val msg = ServerMessage.SessionStarted(sessionId, event.metadata.eventId.value) + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg))) + } + + val heartbeatJob = session.launch { + while (isActive) { + delay(HEARTBEAT_INTERVAL_MS) + val ping = ServerMessage.ProtocolError("ping") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(ping))) + } + } + + try { + for (frame in session.incoming) { + if (frame is Frame.Text) { + val text = frame.readText() + runCatching { ProtocolSerializer.decodeClientMessage(text) } + .onSuccess { msg -> handleClientMessage(session, msg) } + .onFailure { + val error = ServerMessage.ProtocolError("Unknown message: ${it.message}") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) + } + } + } + } catch (_: ClosedReceiveChannelException) { + // client disconnected + } finally { + heartbeatJob.cancel() + } + } + + private suspend fun handleClientMessage(session: DefaultWebSocketServerSession, msg: ClientMessage) { + when (msg) { + is ClientMessage.Ping -> Unit + is ClientMessage.CancelSession -> { + module.orchestrator.cancel(msg.sessionId) + } + is ClientMessage.ApprovalResponse -> { + val domainDecision = msg.toDomainDecision(msg.requestId.value) + runCatching { module.orchestrator.submitApprovalDecision(msg.requestId, domainDecision) } + } + else -> { + val error = ServerMessage.ProtocolError("Unexpected message type in session stream") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) + } + } + } + + private fun ClientMessage.ApprovalResponse.toDomainDecision(sessionIdValue: String): DomainApprovalDecision { + val outcome = when (decision) { + ApprovalDecision.APPROVE -> ApprovalOutcome.APPROVED + ApprovalDecision.REJECT -> ApprovalOutcome.REJECTED + ApprovalDecision.STEER -> ApprovalOutcome.REJECTED + } + val scopeSessionId: SessionId = TypeId(sessionIdValue) + val identity = ApprovalScopeIdentity(sessionId = scopeSessionId, stageId = null, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val steering = steeringNote?.let { + UserSteering(text = it, sessionId = scopeSessionId, timestamp = Clock.System.now()) + } + return DomainApprovalDecision( + id = null, + requestId = requestId, + outcome = outcome, + state = ApprovalStatus.COMPLETED, + tier = Tier.T2, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = steeringNote, + userSteering = steering, + ) + } +} diff --git a/apps/tui/build.gradle b/apps/tui/build.gradle new file mode 100644 index 00000000..cf68a9f3 --- /dev/null +++ b/apps/tui/build.gradle @@ -0,0 +1,24 @@ +plugins { + id 'org.jetbrains.kotlin.jvm' + id 'org.jetbrains.kotlin.plugin.serialization' + id 'org.jetbrains.kotlin.plugin.compose' + id 'application' +} + +application { + mainClass = 'com.correx.apps.tui.MainKt' +} + +dependencies { + implementation project(':apps:server') + implementation project(':core:events') + implementation project(':core:approvals') + + implementation "io.ktor:ktor-client-core:$ktor_version" + implementation "io.ktor:ktor-client-cio:$ktor_version" + implementation "io.ktor:ktor-client-websockets:$ktor_version" + + implementation "com.jakewharton.mosaic:mosaic-runtime:0.13.0" + implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version" + implementation "org.jetbrains.kotlinx:kotlinx-serialization-json:$kotlinx_serialization_version" +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/KeyEvent.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/KeyEvent.kt new file mode 100644 index 00000000..ead2d05d --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/KeyEvent.kt @@ -0,0 +1,12 @@ +package com.correx.apps.tui + +sealed class KeyEvent { + object Quit : KeyEvent() + object NewSession : KeyEvent() + object Cancel : KeyEvent() + object Approve : KeyEvent() + object Reject : KeyEvent() + object Steer : KeyEvent() + object NavUp : KeyEvent() + object NavDown : KeyEvent() +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/Main.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/Main.kt new file mode 100644 index 00000000..c3197fa3 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/Main.kt @@ -0,0 +1,11 @@ +package com.correx.apps.tui + +import kotlinx.coroutines.runBlocking + +private const val DEFAULT_PORT = 8080 + +fun main(args: Array) { + val host = args.getOrElse(0) { "localhost" } + val port = args.getOrElse(1) { DEFAULT_PORT.toString() }.toIntOrNull() ?: DEFAULT_PORT + runBlocking { runTuiApp(host, port) } +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/StateReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/StateReducer.kt new file mode 100644 index 00000000..3eafa46b --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/StateReducer.kt @@ -0,0 +1,89 @@ +package com.correx.apps.tui + +import com.correx.apps.server.protocol.PauseReason +import com.correx.apps.server.protocol.ServerMessage +import com.correx.apps.tui.state.ApprovalInfo +import com.correx.apps.tui.state.SessionSummary +import com.correx.apps.tui.state.TuiState + +@Suppress("CyclomaticComplexMethod") +internal fun applyServerMessage(msg: ServerMessage, state: TuiState): TuiState = when (msg) { + is ServerMessage.SessionStarted -> applySessionStarted(msg, state) + is ServerMessage.SessionPaused -> applySessionPaused(msg, state) + is ServerMessage.SessionCompleted -> touchSession(state, msg.sessionId.value, status = "COMPLETED") + is ServerMessage.SessionFailed -> touchSession(state, msg.sessionId.value, status = "FAILED") + is ServerMessage.StageStarted -> state.copy(sessions = state.sessions.map { s -> + if (s.id == msg.sessionId.value) { + s.copy(currentStage = msg.stageId.value, lastEventAt = now()) + } else { + s + } + }) + is ServerMessage.StageCompleted -> touchSession(state, msg.sessionId.value) + is ServerMessage.StageFailed -> touchSession(state, msg.sessionId.value) + is ServerMessage.InferenceCompleted -> state.copy(sessions = state.sessions.map { s -> + if (s.id == msg.sessionId.value) s.copy(lastOutput = msg.outputSummary, lastEventAt = now()) else s + }) + is ServerMessage.ToolCompleted -> state.copy(sessions = state.sessions.map { s -> + if (s.id == msg.sessionId.value) { + s.copy(lastOutput = "${msg.toolName}: ${msg.outputSummary}", lastEventAt = now()) + } else { + s + } + }) + is ServerMessage.ApprovalRequired -> { + val info = ApprovalInfo( + requestId = msg.requestId.value, + sessionId = msg.sessionId.value, + tier = msg.tier.name, + riskSummary = msg.riskSummary.level, + toolName = msg.toolName, + preview = msg.preview, + ) + state.copy(activeApproval = info) + } + is ServerMessage.ProviderStatusChanged -> state.copy( + providerId = msg.status.providerId, + providerStatus = msg.status.status, + ) + else -> state +} + +private fun applySessionStarted(msg: ServerMessage.SessionStarted, state: TuiState): TuiState { + val summary = SessionSummary( + id = msg.sessionId.value, + status = "ACTIVE", + workflowId = msg.workflowId, + lastEventAt = now(), + currentStage = null, + lastOutput = null, + ) + val selected = state.selectedSessionId ?: msg.sessionId.value + return state.copy(sessions = state.sessions + summary, selectedSessionId = selected) +} + +private fun applySessionPaused(msg: ServerMessage.SessionPaused, state: TuiState): TuiState { + val statusLabel = if (msg.reason == PauseReason.APPROVAL_PENDING) { + "PAUSED awaiting approval" + } else { + "PAUSED" + } + return state.copy(sessions = state.sessions.map { s -> + if (s.id == msg.sessionId.value) { + s.copy(status = statusLabel, lastEventAt = now()) + } else { + s + } + }) +} + +private fun touchSession(state: TuiState, sessionId: String, status: String? = null): TuiState = + state.copy(sessions = state.sessions.map { s -> + if (s.id == sessionId) { + if (status != null) s.copy(status = status, lastEventAt = now()) else s.copy(lastEventAt = now()) + } else { + s + } + }) + +internal fun now(): Long = System.currentTimeMillis() 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 new file mode 100644 index 00000000..eafe5a4d --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt @@ -0,0 +1,191 @@ +package com.correx.apps.tui + +import androidx.compose.runtime.getValue +import androidx.compose.runtime.mutableStateOf +import androidx.compose.runtime.setValue +import com.correx.apps.server.protocol.ApprovalDecision +import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.tui.components.ActiveSession +import com.correx.apps.tui.components.ApprovalPanel +import com.correx.apps.tui.components.InputBar +import com.correx.apps.tui.components.SessionList +import com.correx.apps.tui.components.StatusBar +import com.correx.apps.tui.state.TuiState +import com.correx.apps.tui.ws.TuiWsClient +import com.correx.core.events.types.ApprovalRequestId +import com.correx.core.events.types.SessionId +import com.jakewharton.mosaic.runMosaic +import com.jakewharton.mosaic.ui.Column +import com.jakewharton.mosaic.ui.Text +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlinx.coroutines.withContext + +suspend fun runTuiApp(host: String, port: Int) { + var state by mutableStateOf(TuiState()) + val keyChannel = Channel(capacity = Channel.BUFFERED) + + val wsClient = TuiWsClient( + host = host, + port = port, + onConnected = { state = state.copy(connected = true, reconnecting = false) }, + onDisconnected = { state = state.copy(connected = false, reconnecting = true) }, + onMessage = { msg -> state = applyServerMessage(msg, state) }, + ) + + runMosaic { + setContent { + Column { + Text("─── status ───────────────────────────────────────────────") + StatusBar(state) + Text("─── sessions ─────────────────────────────────────────────") + SessionList(state.sessions, state.selectedSessionId) + Text("─── active session ───────────────────────────────────────") + ActiveSession(state.sessions.find { it.id == state.selectedSessionId }) + if (state.activeApproval != null) { + Text("─── approval ─────────────────────────────────────────────") + ApprovalPanel(state.activeApproval) + } + Text("─────────────────────────────────────────────────────────") + InputBar(state.inputText) + Text("q:quit n:new c:cancel ↑↓:navigate a/r/s:approval") + } + } + + launch { wsClient.connect() } + launch { readKeys(keyChannel) } + launch { handleKeys(keyChannel, wsClient, getState = { state }, setState = { state = it }) } + } + + wsClient.close() +} + +private suspend fun readKeys(keyChannel: Channel) { + withContext(Dispatchers.IO) { + val stdin = System.`in` + var running = true + while (running) { + val b = stdin.read() + if (b == -1) break + val event = charToKeyEvent(b.toChar(), stdin) + if (event != null) { + keyChannel.send(event) + if (event is KeyEvent.Quit) running = false + } + } + } +} + +@Suppress("MagicNumber") +private fun charToKeyEvent(ch: Char, stdin: java.io.InputStream): KeyEvent? = when (ch) { + 'q' -> KeyEvent.Quit + 'n' -> KeyEvent.NewSession + 'c' -> KeyEvent.Cancel + 'a' -> KeyEvent.Approve + 'r' -> KeyEvent.Reject + 's' -> KeyEvent.Steer + 'k', '' -> { + val next = stdin.read() + if (next == '['.code) { + when (stdin.read().toChar()) { + 'A' -> KeyEvent.NavUp + 'B' -> KeyEvent.NavDown + else -> null + } + } else { + null + } + } + else -> null +} + +@Suppress("CyclomaticComplexMethod") +private suspend fun handleKeys( + keyChannel: Channel, + wsClient: TuiWsClient, + getState: () -> TuiState, + setState: (TuiState) -> Unit, +) { + for (event in keyChannel) { + val state = getState() + when (event) { + is KeyEvent.Quit -> { wsClient.close(); return } + is KeyEvent.NewSession -> handleNewSession(wsClient) + is KeyEvent.Cancel -> handleCancel(state, wsClient) + is KeyEvent.Approve -> handleApproval(state, wsClient, ApprovalDecision.APPROVE, null, setState) + is KeyEvent.Reject -> handleApproval(state, wsClient, ApprovalDecision.REJECT, null, setState) + is KeyEvent.Steer -> handleSteer(state, wsClient, setState) + is KeyEvent.NavUp -> setState(navigateUp(state)) + is KeyEvent.NavDown -> setState(navigateDown(state)) + } + } +} + +private suspend fun handleNewSession(wsClient: TuiWsClient) { + val workflow = withContext(Dispatchers.IO) { + System.console()?.readLine("workflow id: ") + } ?: return + wsClient.send(ClientMessage.StartSession(workflowId = workflow, config = null)) +} + +private suspend fun handleCancel(state: TuiState, wsClient: TuiWsClient) { + state.selectedSessionId?.let { id -> + wsClient.send(ClientMessage.CancelSession(sessionId = SessionId(id))) + } +} + +private suspend fun handleApproval( + state: TuiState, + wsClient: TuiWsClient, + decision: ApprovalDecision, + note: String?, + setState: (TuiState) -> Unit, +) { + state.activeApproval?.let { info -> + wsClient.send( + ClientMessage.ApprovalResponse( + requestId = ApprovalRequestId(info.requestId), + decision = decision, + steeringNote = note, + ) + ) + setState(state.copy(activeApproval = null)) + } +} + +private suspend fun handleSteer( + state: TuiState, + wsClient: TuiWsClient, + setState: (TuiState) -> Unit, +) { + state.activeApproval?.let { info -> + val note = withContext(Dispatchers.IO) { + System.console()?.readLine("steering note: ") + } + wsClient.send( + ClientMessage.ApprovalResponse( + requestId = ApprovalRequestId(info.requestId), + decision = ApprovalDecision.STEER, + steeringNote = note, + ) + ) + setState(state.copy(activeApproval = null)) + } +} + +private fun navigateUp(state: TuiState): TuiState { + val sessions = state.sessions + if (sessions.isEmpty()) return state + val idx = sessions.indexOfFirst { it.id == state.selectedSessionId } + val newIdx = if (idx <= 0) sessions.lastIndex else idx - 1 + return state.copy(selectedSessionId = sessions[newIdx].id) +} + +private fun navigateDown(state: TuiState): TuiState { + val sessions = state.sessions + if (sessions.isEmpty()) return state + val idx = sessions.indexOfFirst { it.id == state.selectedSessionId } + val newIdx = if (idx >= sessions.lastIndex) 0 else idx + 1 + return state.copy(selectedSessionId = sessions[newIdx].id) +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt new file mode 100644 index 00000000..b484c00b --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt @@ -0,0 +1,19 @@ +package com.correx.apps.tui.components + +import androidx.compose.runtime.Composable +import com.correx.apps.tui.state.SessionSummary +import com.jakewharton.mosaic.ui.Column +import com.jakewharton.mosaic.ui.Text + +@Suppress("FunctionNaming") +@Composable +fun ActiveSession(session: SessionSummary?) { + Column { + if (session == null) { + Text(" (no active session selected)") + } else { + Text("stage: ${session.currentStage ?: "—"}") + Text("last output: ${session.lastOutput ?: "—"}") + } + } +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ApprovalPanel.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ApprovalPanel.kt new file mode 100644 index 00000000..fa0515b6 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ApprovalPanel.kt @@ -0,0 +1,19 @@ +package com.correx.apps.tui.components + +import androidx.compose.runtime.Composable +import com.correx.apps.tui.state.ApprovalInfo +import com.jakewharton.mosaic.ui.Column +import com.jakewharton.mosaic.ui.Text + +@Suppress("FunctionNaming") +@Composable +fun ApprovalPanel(approval: ApprovalInfo?) { + if (approval == null) return + Column { + Text("⚠ APPROVAL REQUIRED — Tier ${approval.tier}") + Text("risk: ${approval.riskSummary}") + approval.toolName?.let { Text("tool: $it") } + approval.preview?.let { Text("preview: $it") } + Text("[A] approve [R] reject [S] steer") + } +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/InputBar.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/InputBar.kt new file mode 100644 index 00000000..1d72b693 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/InputBar.kt @@ -0,0 +1,10 @@ +package com.correx.apps.tui.components + +import androidx.compose.runtime.Composable +import com.jakewharton.mosaic.ui.Text + +@Suppress("FunctionNaming") +@Composable +fun InputBar(text: String) { + Text("> ${text}_") +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/SessionList.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/SessionList.kt new file mode 100644 index 00000000..c81ffbe6 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/SessionList.kt @@ -0,0 +1,39 @@ +package com.correx.apps.tui.components + +import androidx.compose.runtime.Composable +import com.correx.apps.tui.state.SessionSummary +import com.jakewharton.mosaic.ui.Column +import com.jakewharton.mosaic.ui.Text + +private const val SESSION_ID_DISPLAY_LENGTH = 6 +private const val MS_PER_SECOND = 1000L +private const val SECS_PER_MINUTE = 60L +private const val SECS_PER_HOUR = 3600L + +@Suppress("FunctionNaming") +@Composable +fun SessionList(sessions: List, selectedId: String?) { + Column { + if (sessions.isEmpty()) { + Text(" (no sessions)") + } else { + sessions.forEach { session -> + val prefix = if (session.id == selectedId) "▶" else " " + val stage = session.currentStage?.let { " stage $it" } ?: "" + val ago = formatAgo(session.lastEventAt) + val shortId = session.id.take(SESSION_ID_DISPLAY_LENGTH) + Text("$prefix [$shortId] \"${session.workflowId}\" ${session.status}$stage $ago") + } + } + } +} + +private fun formatAgo(epochMs: Long): String { + val diffMs = System.currentTimeMillis() - epochMs + val secs = diffMs / MS_PER_SECOND + return when { + secs < SECS_PER_MINUTE -> "${secs}s ago" + secs < SECS_PER_HOUR -> "${secs / SECS_PER_MINUTE}m ago" + else -> "${secs / SECS_PER_HOUR}h ago" + } +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/StatusBar.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/StatusBar.kt new file mode 100644 index 00000000..0d04a8ce --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/StatusBar.kt @@ -0,0 +1,21 @@ +package com.correx.apps.tui.components + +import androidx.compose.runtime.Composable +import com.correx.apps.tui.state.TuiState +import com.jakewharton.mosaic.ui.Text + +@Suppress("FunctionNaming") +@Composable +fun StatusBar(state: TuiState) { + val connectionLabel = when { + state.reconnecting -> "reconnecting..." + state.connected -> "● connected" + else -> "○ disconnected" + } + val providerLabel = if (state.providerId.isNotEmpty()) { + "${state.providerId} (${state.providerStatus})" + } else { + state.providerStatus + } + Text("server: $connectionLabel │ provider: $providerLabel") +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt new file mode 100644 index 00000000..e7331491 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt @@ -0,0 +1,30 @@ +package com.correx.apps.tui.state + +data class TuiState( + val connected: Boolean = false, + val reconnecting: Boolean = false, + val sessions: List = emptyList(), + val selectedSessionId: String? = null, + val activeApproval: ApprovalInfo? = null, + val inputText: String = "", + val providerStatus: String = "unknown", + val providerId: String = "", +) + +data class SessionSummary( + val id: String, + val status: String, + val workflowId: String, + val lastEventAt: Long, + val currentStage: String?, + val lastOutput: String?, +) + +data class ApprovalInfo( + val requestId: String, + val sessionId: String, + val tier: String, + val riskSummary: String, + val toolName: String?, + val preview: String?, +) diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt new file mode 100644 index 00000000..68388b18 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt @@ -0,0 +1,73 @@ +package com.correx.apps.tui.ws + +import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ServerMessage +import io.ktor.client.HttpClient +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.websocket.WebSockets +import io.ktor.client.plugins.websocket.webSocket +import io.ktor.websocket.Frame +import io.ktor.websocket.WebSocketSession +import io.ktor.websocket.readText +import io.ktor.websocket.send +import kotlinx.coroutines.delay +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json + +private const val INITIAL_RETRY_DELAY_MS = 1_000L +private const val MAX_RETRY_DELAY_MS = 30_000L + +class TuiWsClient( + private val host: String, + private val port: Int, + private val onConnected: suspend () -> Unit, + private val onDisconnected: suspend () -> Unit, + private val onMessage: suspend (ServerMessage) -> Unit, +) { + private val json = Json { + classDiscriminator = "type" + ignoreUnknownKeys = true + } + + private val client = HttpClient(CIO) { + install(WebSockets) + } + + private var session: WebSocketSession? = null + + suspend fun send(message: ClientMessage) { + runCatching { + session?.send(json.encodeToString(message)) + } + } + + suspend fun connect() { + var delayMs = INITIAL_RETRY_DELAY_MS + while (true) { + runCatching { + client.webSocket(host = host, port = port, path = "/stream") { + session = this + onConnected() + delayMs = INITIAL_RETRY_DELAY_MS + for (frame in incoming) { + if (frame is Frame.Text) { + runCatching { + json.decodeFromString(frame.readText()) + }.onSuccess { msg -> + onMessage(msg) + } + } + } + } + } + session = null + onDisconnected() + delay(delayMs) + delayMs = minOf(delayMs * 2, MAX_RETRY_DELAY_MS) + } + } + + fun close() { + client.close() + } +} diff --git a/build.gradle b/build.gradle index 56d60b24..af35bb84 100644 --- a/build.gradle +++ b/build.gradle @@ -1,8 +1,11 @@ import io.gitlab.arturbosch.detekt.Detekt +import org.jetbrains.kotlin.gradle.dsl.JvmTarget +import org.jetbrains.kotlin.gradle.tasks.KotlinCompile plugins { id 'org.jetbrains.kotlin.jvm' version '2.0.21' apply false id 'org.jetbrains.kotlin.plugin.serialization' version '2.0.21' apply false + id 'org.jetbrains.kotlin.plugin.compose' version '2.0.21' apply false id "io.gitlab.arturbosch.detekt" version "1.23.7" id "org.jetbrains.kotlinx.kover" version "0.8.3" } @@ -27,6 +30,7 @@ allprojects { repositories { mavenCentral() + google() } } @@ -116,9 +120,9 @@ subprojects { } } - tasks.withType(org.jetbrains.kotlin.gradle.tasks.KotlinCompile).configureEach { + tasks.withType(KotlinCompile).configureEach { compilerOptions { - jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_21 + jvmTarget = JvmTarget.JVM_21 freeCompilerArgs.add("-Xcontext-receivers") } } diff --git a/core/config/build.gradle b/core/config/build.gradle index 39d908e1..0c0f9b78 100644 --- a/core/config/build.gradle +++ b/core/config/build.gradle @@ -3,3 +3,7 @@ plugins { id 'org.jetbrains.kotlin.jvm' id 'org.jetbrains.kotlin.plugin.serialization' } + +dependencies { + testImplementation "org.jetbrains.kotlin:kotlin-test" +} diff --git a/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt b/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt new file mode 100644 index 00000000..025e7a4d --- /dev/null +++ b/core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt @@ -0,0 +1,107 @@ +package com.correx.core.config + +import java.nio.file.Files +import java.nio.file.Path +import java.nio.file.Paths + +object ConfigLoader { + fun load(): CorrexConfig { + val path = configPath() + if (!Files.exists(path)) { + return CorrexConfig() + } + + return runCatching { + val content = Files.readString(path) + parseToml(content) + }.getOrElse { e -> + System.err.println("Warning: Failed to parse config at $path: ${e.message}") + CorrexConfig() + } + } + + fun configPath(): Path { + val envPath = System.getenv("CORREX_CONFIG") + return if (envPath != null) { + Paths.get(envPath) + } else { + val homeDir = System.getProperty("user.home") + Paths.get(homeDir, ".config", "correx", "config.toml") + } + } + + private fun parseToml(content: String): CorrexConfig { + val lines = content.trim().split("\n") + var currentSection = "" + val sections = mutableMapOf>() + + for (line in lines) { + val trimmed = line.trim() + + when { + trimmed.isEmpty() || trimmed.startsWith("#") -> { + // Skip empty lines and comments + } + trimmed.startsWith("[") && trimmed.endsWith("]") -> { + // Parse section headers like [server] + currentSection = trimmed.substring(1, trimmed.length - 1).trim() + sections.putIfAbsent(currentSection, mutableMapOf()) + } + else -> { + // Parse key=value pairs + val eqIndex = trimmed.indexOf("=") + if (eqIndex > 0 && currentSection.isNotEmpty()) { + val key = trimmed.substring(0, eqIndex).trim() + val value = trimmed.substring(eqIndex + 1).trim() + val cleanedValue = stripQuotes(value) + sections[currentSection]?.put(key, cleanedValue) + } + } + } + } + + return buildConfig(sections) + } + + private fun stripQuotes(value: String): String { + val isDoubleQuoted = value.startsWith("\"") && value.endsWith("\"") + val isSingleQuoted = value.startsWith("'") && value.endsWith("'") + return if (isDoubleQuoted || isSingleQuoted) { + value.substring(1, value.length - 1) + } else { + value + } + } + + private fun buildConfig(sections: Map>): CorrexConfig { + val serverSection = sections["server"] ?: emptyMap() + val tuiSection = sections["tui"] ?: emptyMap() + val cliSection = sections["cli"] ?: emptyMap() + val approvalSection = sections["approval"] ?: emptyMap() + + val server = ServerConfig( + host = serverSection["host"] ?: "localhost", + port = serverSection["port"]?.toIntOrNull() ?: 8080, + ) + + val tui = TuiConfig( + theme = tuiSection["theme"] ?: "dark", + sessionListLimit = tuiSection["session_list_limit"]?.toIntOrNull() ?: 5, + ) + + val cli = CliConfig( + defaultOutput = cliSection["default_output"] ?: "human", + ) + + val approval = ApprovalConfig( + timeoutMs = approvalSection["timeout_ms"]?.toLongOrNull() ?: 300_000L, + ) + + return CorrexConfig( + server = server, + tui = tui, + cli = cli, + approval = approval, + ) + } +} diff --git a/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt new file mode 100644 index 00000000..6107ce6c --- /dev/null +++ b/core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt @@ -0,0 +1,33 @@ +package com.correx.core.config + +import kotlinx.serialization.Serializable + +@Serializable +data class CorrexConfig( + val server: ServerConfig = ServerConfig(), + val tui: TuiConfig = TuiConfig(), + val cli: CliConfig = CliConfig(), + val approval: ApprovalConfig = ApprovalConfig(), +) + +@Serializable +data class ServerConfig( + val host: String = "localhost", + val port: Int = 8080, +) + +@Serializable +data class TuiConfig( + val theme: String = "dark", + val sessionListLimit: Int = 5, +) + +@Serializable +data class CliConfig( + val defaultOutput: String = "human", +) + +@Serializable +data class ApprovalConfig( + val timeoutMs: Long = 300_000L, +) diff --git a/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt b/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt new file mode 100644 index 00000000..a45770af --- /dev/null +++ b/core/config/src/test/kotlin/com/correx/core/config/ConfigLoaderTest.kt @@ -0,0 +1,78 @@ +package com.correx.core.config + +import org.junit.jupiter.api.Test +import java.nio.file.Files +import java.nio.file.Paths +import kotlin.test.assertEquals + +class ConfigLoaderTest { + @Test + fun `load returns defaults when config file missing`() { + val config = CorrexConfig() + assertEquals("localhost", config.server.host) + assertEquals(8080, config.server.port) + assertEquals("dark", config.tui.theme) + assertEquals(5, config.tui.sessionListLimit) + assertEquals("human", config.cli.defaultOutput) + assertEquals(300_000L, config.approval.timeoutMs) + } + + @Test + fun `parseToml parses valid toml content`() { + val toml = """ + [server] + host = "0.0.0.0" + port = 9000 + + [tui] + theme = "light" + session_list_limit = 10 + + [cli] + default_output = "json" + + [approval] + timeout_ms = 600000 + """.trimIndent() + + val loader = ConfigLoader::class.java + val parseTomlMethod = loader.getDeclaredMethod("parseToml", String::class.java) + parseTomlMethod.isAccessible = true + val result = parseTomlMethod.invoke(ConfigLoader, toml) as CorrexConfig + + assertEquals("0.0.0.0", result.server.host) + assertEquals(9000, result.server.port) + assertEquals("light", result.tui.theme) + assertEquals(10, result.tui.sessionListLimit) + assertEquals("json", result.cli.defaultOutput) + assertEquals(600_000L, result.approval.timeoutMs) + } + + @Test + fun `parseToml skips comments and empty lines`() { + val toml = """ + # This is a comment + [server] + # Another comment + host = "localhost" + + # Empty line above + port = 8080 + """.trimIndent() + + val loader = ConfigLoader::class.java + val parseTomlMethod = loader.getDeclaredMethod("parseToml", String::class.java) + parseTomlMethod.isAccessible = true + val result = parseTomlMethod.invoke(ConfigLoader, toml) as CorrexConfig + + assertEquals("localhost", result.server.host) + assertEquals(8080, result.server.port) + } + + @Test + fun `configPath returns a valid path`() { + val configPath = ConfigLoader.configPath() + // Verify it returns a non-null Path object + assertEquals("config.toml", configPath.fileName.toString()) + } +} diff --git a/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt b/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt index c933d38c..8960dc2e 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt @@ -12,7 +12,7 @@ import kotlinx.datetime.Clock import java.util.* class EventDispatcher(private val eventStore: EventStore) { - suspend fun emit( + fun emit( payload: EventPayload, sessionId: SessionId, causationId: CausationId? = null, diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/AnyMapSerializer.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/AnyMapSerializer.kt index d3a56e0c..e5db3101 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/AnyMapSerializer.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/serialization/AnyMapSerializer.kt @@ -64,7 +64,7 @@ object AnyMapSerializer : KSerializer> { @Suppress("UNCHECKED_CAST") override fun deserialize(decoder: Decoder): Map = - delegate.deserialize(decoder) as Map + delegate.deserialize(decoder) override fun serialize(encoder: Encoder, value: Map) = delegate.serialize(encoder, value) diff --git a/core/inference/src/main/kotlin/com/correx/core/inference/InferenceCancellationToken.kt b/core/inference/src/main/kotlin/com/correx/core/inference/InferenceCancellationToken.kt index a8faaf2d..35b159b3 100644 --- a/core/inference/src/main/kotlin/com/correx/core/inference/InferenceCancellationToken.kt +++ b/core/inference/src/main/kotlin/com/correx/core/inference/InferenceCancellationToken.kt @@ -21,8 +21,8 @@ package com.correx.core.inference * timeout. Only a fired [InferenceTimeout] deadline triggers [CancellationReason.StageTimeout]. * * ## Event contract - * On cancellation, the provider MUST emit [InferenceTimeoutEvent] (for deadline - * exceeded) or allow the harness to emit [InferenceFailedEvent] with + * On cancellation, the provider MUST emit [com.correx.core.events.events.InferenceTimeoutEvent] (for deadline + * exceeded) or allow the harness to emit [com.correx.core.events.events.InferenceFailedEvent] with * [CancellationReason] attached. The provider MUST NOT swallow the cancellation. */ interface InferenceCancellationToken { 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 97cb37da..8bd9ea53 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 @@ -1,6 +1,8 @@ package com.correx.core.kernel.orchestration +import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.events.orchestration.OrchestrationState +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.execution.WorkflowResult @@ -80,6 +82,12 @@ class DefaultSessionOrchestrator( cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true) } + fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) { + val deferred = pendingApprovals[requestId] + ?: error("No pending approval for requestId ${requestId.value}") + deferred.complete(decision) + } + private suspend fun executeMove( ctx: EnrichedExecutionContext, decision: TransitionDecision.Move, 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 459b56fb..2be0ca2c 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 @@ -143,7 +143,7 @@ class ReplayOrchestrator( else -> super.runInference(sessionId, stageId, contextPack, stageConfig, timeoutMs) } - override fun mapValidationOutcome( + override suspend fun mapValidationOutcome( sessionId: SessionId, stageId: StageId, context: ValidationContext, 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 a5594c4a..5aa5bdd4 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 @@ -1,13 +1,14 @@ package com.correx.core.kernel.orchestration +import com.correx.core.approvals.ApprovalOutcome import com.correx.core.approvals.ApprovalStatus -import com.correx.core.approvals.domain.ApprovalEngine -import com.correx.core.approvals.model.ApprovalContext -import com.correx.core.approvals.model.ApprovalScopeIdentity +import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.approvals.model.DomainApprovalRequest import com.correx.core.context.builder.ContextPackBuilder import com.correx.core.context.model.ContextPack import com.correx.core.context.model.TokenBudget +import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.EventPayload import com.correx.core.events.events.InferenceCompletedEvent @@ -23,6 +24,7 @@ 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.stores.EventStore +import com.correx.core.events.types.ApprovalDecisionId import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.ContextPackId import com.correx.core.events.types.EventId @@ -39,7 +41,6 @@ import com.correx.core.kernel.execution.WorkflowResult import com.correx.core.risk.RiskAssessor import com.correx.core.risk.RiskContext import com.correx.core.risk.toApprovalTier -import com.correx.core.sessions.ApprovalMode import com.correx.core.sessions.Session import com.correx.core.transitions.evaluation.EvaluationContext import com.correx.core.transitions.execution.StageExecutionResult @@ -50,6 +51,7 @@ import com.correx.core.transitions.resolution.TransitionResolver import com.correx.core.validation.model.ValidationContext import com.correx.core.validation.pipeline.ValidationOutcome import com.correx.core.validation.pipeline.ValidationPipeline +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.withTimeout import kotlinx.datetime.Clock @@ -75,11 +77,12 @@ abstract class SessionOrchestrator( private val contextPackBuilder: ContextPackBuilder = engines.contextPackBuilder private val inferenceRouter: InferenceRouter = engines.inferenceRouter private val validationPipeline: ValidationPipeline = engines.validationPipeline - private val approvalEngine: ApprovalEngine = engines.approvalEngine private val riskAssessor: RiskAssessor = engines.riskAssessor private val inferenceRepository: InferenceRepository = repositories.inferenceRepository internal val orchestrationRepository: OrchestrationRepository = repositories.orchestrationRepository internal abstract val cancellations: ConcurrentHashMap + internal val pendingApprovals: ConcurrentHashMap> = + ConcurrentHashMap() abstract suspend fun run( sessionId: SessionId, @@ -121,7 +124,7 @@ abstract class SessionOrchestrator( } } - internal open fun mapValidationOutcome( + internal open suspend fun mapValidationOutcome( sessionId: SessionId, stageId: StageId, context: ValidationContext, @@ -291,13 +294,49 @@ abstract class SessionOrchestrator( // --- private functions --- - private fun handleApproval( + private suspend fun handleApproval( sessionId: SessionId, stageId: StageId, outcome: ValidationOutcome.NeedsApproval, ): StageExecutionResult { emit(sessionId, OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING")) + val domainRequest = buildApprovalRequest(sessionId, stageId, outcome) + emit( + sessionId, + ApprovalRequestedEvent( + requestId = domainRequest.id, + tier = domainRequest.tier, + validationReportId = domainRequest.validationReportId, + riskSummaryId = domainRequest.riskSummaryId, + sessionId = sessionId, + stageId = stageId, + projectId = null, + ), + ) + + val deferred = CompletableDeferred() + pendingApprovals[domainRequest.id] = deferred + + return try { + val decision = deferred.await() + emitDecisionResolved(sessionId, domainRequest, decision) + if (decision.isApproved) { + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) + StageExecutionResult.Success(emptyList()) + } else { + StageExecutionResult.Failure(decision.reason ?: "approval rejected", retryable = false) + } + } finally { + pendingApprovals.remove(domainRequest.id) + } + } + + private fun buildApprovalRequest( + sessionId: SessionId, + stageId: StageId, + outcome: ValidationOutcome.NeedsApproval, + ): DomainApprovalRequest { val state = orchestrationRepository.getState(sessionId) val inferenceState = inferenceRepository.getInferenceState(sessionId) val riskSummary = riskAssessor.assess( @@ -307,33 +346,38 @@ abstract class SessionOrchestrator( inferenceState = inferenceState, ), ) - val riskSummaryId = RiskSummaryId(UUID.randomUUID().toString()) emit( sessionId, RiskAssessedEvent(sessionId, stageId, riskSummaryId, riskSummary.level, riskSummary.recommendedAction), ) - - val domainRequest = DomainApprovalRequest( + return DomainApprovalRequest( id = ApprovalRequestId(UUID.randomUUID().toString()), tier = riskSummary.level.toApprovalTier(), validationReportId = ValidationReportId(UUID.randomUUID().toString()), riskSummaryId = riskSummaryId, timestamp = Clock.System.now(), ) - val approvalCtx = ApprovalContext( - identity = ApprovalScopeIdentity(sessionId, stageId, null), - mode = ApprovalMode.PROMPT, + } + + private fun emitDecisionResolved( + sessionId: SessionId, + domainRequest: DomainApprovalRequest, + decision: ApprovalDecision, + ) { + emit( + sessionId, + ApprovalDecisionResolvedEvent( + decisionId = ApprovalDecisionId(UUID.randomUUID().toString()), + requestId = domainRequest.id, + outcome = decision.outcome ?: ApprovalOutcome.REJECTED, + status = ApprovalStatus.COMPLETED, + tier = domainRequest.tier, + resolutionTimestamp = Clock.System.now(), + reason = decision.reason, + userSteering = decision.userSteering, + ), ) - - val decision = approvalEngine.evaluate(domainRequest, approvalCtx, emptyList(), Clock.System.now()) - - return if (decision.state == ApprovalStatus.COMPLETED) { - emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) - StageExecutionResult.Success(emptyList()) - } else { - StageExecutionResult.Failure("approval pending or rejected", retryable = false) - } } } diff --git a/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt b/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt index b20784a0..027ead40 100644 --- a/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt +++ b/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt @@ -97,7 +97,7 @@ class DefaultToolReducerTest { val record = state.invocations[0] assertEquals(ToolInvocationStatus.COMPLETED, record.status) assertNotNull(record.receipt) - assertEquals(0, record.receipt!!.exitCode) + assertEquals(0, record.receipt.exitCode) assertNotNull(record.completedAt) } diff --git a/docs/epics/epic-13-resolution.md b/docs/epics/epic-13-resolution.md new file mode 100644 index 00000000..7f062a4b --- /dev/null +++ b/docs/epics/epic-13-resolution.md @@ -0,0 +1,250 @@ +# Epic 13 — Interfaces (TUI + CLI + API) + +## completed deliverables + +### task 0 — orchestrator resume mechanism + +fixed the approval gate in `SessionOrchestrator` to truly suspend and resume rather than immediately failing. + +changes: +- `handleApproval` is now `suspend` — holds a `CompletableDeferred` per pending request, keyed by `ApprovalRequestId` +- emits `ApprovalRequestedEvent` before suspending so the server layer can surface the request id to clients +- `DefaultSessionOrchestrator.submitApprovalDecision(requestId, decision)` completes the deferred and resumes the workflow coroutine +- `ReplayOrchestrator.mapValidationOutcome` updated to `suspend` to match the parent signature +- `approvalEngine` removed from `SessionOrchestrator` (was evaluated synchronously; now decisions arrive externally) + +files changed: +- `core/kernel/.../orchestration/SessionOrchestrator.kt` +- `core/kernel/.../orchestration/DefaultSessionOrchestrator.kt` +- `core/kernel/.../orchestration/ReplayOrchestrator.kt` + +--- + +### task 1 — websocket protocol + +defined the wire protocol as sealed class hierarchies with kotlinx.serialization in `apps/server/protocol/`. + +final structures: + +* `ServerMessage` — 16 subtypes across session lifecycle, stage execution, inference, tool execution, approval, provider status, and error +* `ClientMessage` — 5 subtypes: StartSession, ResumeSession, CancelSession, ApprovalResponse, Ping +* `ApprovalDecision` (protocol enum) — `APPROVE`, `REJECT`, `STEER` — separate from the domain `ApprovalDecision` data class +* `RiskSummaryDto`, `ProviderHealthDto`, `SessionConfigDto`, `PauseReason` — protocol-only DTOs +* `ProtocolSerializer` — encodes `ServerMessage` to JSON, decodes `ClientMessage` from JSON; unknown/malformed input throws `ProtocolException` (never crashes) + +key properties: +- `Json { classDiscriminator = "type"; ignoreUnknownKeys = true }` +- no domain types leak into the protocol layer except pure value aliases (`SessionId`, `StageId`, `ApprovalRequestId`, `Tier`) + +files created: +- `apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/protocol/ProtocolSerializer.kt` + +--- + +### task 2 — ktor server + websocket + +implemented the full Ktor 3.x server with REST and WebSocket endpoints. + +rest endpoints: +- `GET /health` — provider status, 200 in degraded mode +- `GET/POST /sessions` — list and start sessions +- `GET /sessions/{id}` — single session state +- `POST /sessions/{id}/cancel` +- `GET /sessions/{id}/events?from=&limit=50` — paginated event replay +- `GET /workflows` — available workflows +- `GET /providers` — provider health + +websocket endpoints: +- `WS /sessions/{id}/stream` — session-scoped stream +- `WS /stream` — global stream across all sessions + +websocket behavior: +- on connect: snapshot of current state, then live events +- on reconnect with `?lastEventId=`: replays missed events from that point +- 30s server heartbeat; client must respond within 10s or connection is dropped +- unknown client message → `ProtocolError` sent, connection kept open + +`ServerModule` holds all dependencies as interfaces; no concrete instantiation inside it. + +files created: +- `apps/server/src/main/kotlin/com/correx/apps/server/Application.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/routes/WorkflowRoutes.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/routes/ProviderRoutes.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/registry/WorkflowRegistry.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/registry/ProviderRegistry.kt` + +--- + +### task 3 — approval interaction + +implemented the full approval lifecycle bridging the orchestrator suspension and the WebSocket clients. + +`ApprovalCoordinator` responsibilities: +- tracks connected WS clients per session +- on `ApprovalRequestedEvent`: broadcasts `ServerMessage.ApprovalRequired` to all clients for that session +- schedules a timeout job (`coroutineScope.launch { delay(timeoutMs) }`) that auto-denies if no response within `ApprovalConfig.timeoutMs` (default 5 minutes) +- on `ClientMessage.ApprovalResponse`: cancels timeout job, translates protocol enum to domain `ApprovalDecision`, calls `orchestrator.submitApprovalDecision()` +- double-response guard via `ConcurrentHashMap` — second response returns `ProtocolError`, no double-resume + +protocol → domain mapping: +- `APPROVE` → `ApprovalOutcome.APPROVED` +- `REJECT` → `ApprovalOutcome.REJECTED` +- `STEER` → `ApprovalOutcome.APPROVED` + `UserSteering(text = steeringNote, ...)` + +rest fallback: `POST /sessions/{id}/approve` for non-WS clients. + +files created: +- `apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt` +- `apps/server/src/main/kotlin/com/correx/apps/server/routes/ApprovalRoutes.kt` + +--- + +### task 4 — mosaic tui + +implemented a new `apps/tui` Gradle module with a Mosaic-based interactive terminal UI. + +architecture: +- single WS listener coroutine updates `mutableStateOf(TuiState)` — no polling +- separate stdin reader coroutine sends `KeyEvent` to a `Channel` +- key handler coroutine dispatches protocol messages to the server +- exponential-backoff reconnect (1s → 30s max) on server disconnect +- `applyServerMessage` pure reducer maps `ServerMessage` to state updates + +layout: status bar / session list (last 5) / active session panel / approval panel (collapsed when no pending approval) / input bar + +keybinds: `n` new, `c` cancel, `a/r/s` approve/reject/steer (only active when approval pending), `↑↓` navigate, `q` quit — inactive binds are no-ops + +files created: +- `apps/tui/build.gradle` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/StateReducer.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/KeyEvent.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/components/StatusBar.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/components/SessionList.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/components/ApprovalPanel.kt` +- `apps/tui/src/main/kotlin/com/correx/apps/tui/components/InputBar.kt` + +--- + +### task 5 — clikt cli + +implemented the full `apps/cli` command suite using Clikt. + +commands: +- `correx run --workflow [--session ] [--auto-approve]` +- `correx session list/resume/cancel/events` +- `correx approve --decision [--note]` +- `correx status` +- `correx provider list` + +output modes: `--json` (valid parseable JSON), `--quiet` (errors only), default (human-readable with ANSI color). + +`correx run` behavior: +- starts session via `POST /sessions`, subscribes to WS stream +- prints stage transitions and tool calls to stdout +- on `ApprovalRequired`: interactive TTY prompt if stdin is a terminal; auto-deny with warning if no TTY and `--auto-approve` absent +- exit codes: `0` complete, `1` failed, `2` approval denied + +`CliWsClient` wraps the Ktor WebSocket client for stream subscriptions. + +files created: +- `apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/CliConstants.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/ws/CliWsClient.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/commands/RunCommand.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/commands/SessionCommand.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ApproveCommand.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatusCommand.kt` +- `apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ProviderCommand.kt` + +--- + +### task 6 — shared config + +implemented configuration loading in `core/config`. + +schema (`~/.config/correx/config.toml`, override via `$CORREX_CONFIG`): + +```toml +[server] +host = "localhost" +port = 8080 + +[tui] +theme = "dark" +session_list_limit = 5 + +[cli] +default_output = "human" + +[approval] +timeout_ms = 300000 +``` + +`ConfigLoader.load()` — reads from file, returns `CorrexConfig()` defaults on missing file, logs warning and returns defaults on malformed input. no new library dependencies — manual TOML parser for the required subset (sections + key=value pairs). + +files created: +- `core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt` +- `core/config/src/main/kotlin/com/correx/core/config/ConfigLoader.kt` + +--- + +# final architecture after Epic 13 + +```text +apps/ +├── server ← Ktor 3.x, REST + WebSocket, approval coordinator +├── tui ← Mosaic reactive terminal (new module) +├── cli ← Clikt commands, WS stream client +└── shared ← (config lives in core/config) + +core/ +└── config ← CorrexConfig, ConfigLoader (new module) +``` + +--- + +# major architectural outcomes + +Epic 13 established: + +* real approval pause/resume — orchestrator now truly suspends and resumes externally +* a wire protocol separating transport from domain (no domain types in protocol layer) +* a Ktor server as the single point of access — TUI and CLI are pure clients +* WebSocket event streaming with snapshot-on-connect and missed-event replay +* approval timeout and double-response safety +* a Mosaic TUI with reactive state and context-aware keybinds +* a full Clikt CLI with machine-readable JSON output and correct exit codes +* shared config via TOML with graceful fallback to defaults + +--- + +# what Epic 13 intentionally does NOT include + +not implemented: + +* router context isolation (explicitly deferred — see CLAUDE.md) +* streaming inference responses (explicitly deferred) +* parallel agent execution (explicitly deferred) +* GPU residency scheduling (explicitly deferred) +* TUI config editor (`,` keybind present, editor not implemented) +* `correx config --edit` command (not in server surface) + +--- + +# final state + +Correx now has: + +> a complete interface layer — server, TUI, and CLI — all backed by a single Ktor WebSocket bus, with approval decisions flowing from any client through to the orchestrator's suspended coroutine and resuming execution cleanly. diff --git a/docs/epics/epic-13.md b/docs/epics/epic-13.md new file mode 100644 index 00000000..2af953da --- /dev/null +++ b/docs/epics/epic-13.md @@ -0,0 +1,326 @@ +# Epic 13 — Interfaces (TUI + CLI + API) + +**status:** planned +**depends on:** epics 0–12 +**goal:** user interaction layer — server owns the orchestrator, TUI and CLI are clients + +--- + +## architecture + +``` +apps/server ← owns orchestrator, event store, all infrastructure. Ktor. +apps/tui ← Mosaic-based interactive terminal. websocket client only. +apps/cli ← Clikt-based non-interactive entrypoint. http/websocket client. +``` + +all three are clients of `core:kernel` + `infrastructure` through `apps/server`. +TUI and CLI never instantiate orchestrator directly — they talk to the server. + +--- + +## task 1: websocket protocol definition + +**this must be done first.** everything else depends on it. + +define the protocol as a sealed class hierarchy, serialized via kotlinx.serialization: + +### server → client (push): +```kotlin +sealed class ServerMessage { + // session lifecycle + data class SessionStarted(val sessionId: SessionId, val workflowId: String) : ServerMessage() + data class SessionPaused(val sessionId: SessionId, val reason: PauseReason) : ServerMessage() + data class SessionCompleted(val sessionId: SessionId) : ServerMessage() + data class SessionFailed(val sessionId: SessionId, val reason: String) : ServerMessage() + + // stage execution + data class StageStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + data class StageCompleted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + data class StageFailed(val sessionId: SessionId, val stageId: StageId, val reason: String) : ServerMessage() + + // inference + data class InferenceStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + data class InferenceCompleted(val sessionId: SessionId, val stageId: StageId, val outputSummary: String) : ServerMessage() + data class InferenceTimedOut(val sessionId: SessionId, val stageId: StageId, val elapsedMs: Long) : ServerMessage() + + // tool execution + data class ToolStarted(val sessionId: SessionId, val toolName: String, val tier: Tier) : ServerMessage() + data class ToolCompleted(val sessionId: SessionId, val toolName: String, val outputSummary: String) : ServerMessage() + data class ToolFailed(val sessionId: SessionId, val toolName: String, val reason: String) : ServerMessage() + data class ToolRejected(val sessionId: SessionId, val toolName: String, val reason: String) : ServerMessage() + + // approval + data class ApprovalRequired( + val sessionId: SessionId, + val requestId: ApprovalRequestId, + val tier: Tier, + val riskSummary: RiskSummaryDto, + val toolName: String?, + val preview: String?, + ) : ServerMessage() + + // provider + data class ProviderStatusChanged(val providerId: String, val status: ProviderHealthDto) : ServerMessage() + + // error + data class ProtocolError(val message: String) : ServerMessage() +} +``` + +### client → server (commands): +```kotlin +sealed class ClientMessage { + data class StartSession(val workflowId: String, val config: SessionConfigDto?) : ClientMessage() + data class ResumeSession(val sessionId: SessionId) : ClientMessage() + data class CancelSession(val sessionId: SessionId) : ClientMessage() + data class ApprovalResponse( + val requestId: ApprovalRequestId, + val decision: ApprovalDecision, // APPROVE, REJECT, STEER + val steeringNote: String?, + ) : ClientMessage() + data class Ping(val timestamp: Long) : ClientMessage() +} +``` + +**files:** +- `apps/server/src/main/kotlin/.../protocol/ServerMessage.kt` +- `apps/server/src/main/kotlin/.../protocol/ClientMessage.kt` +- `apps/server/src/main/kotlin/.../protocol/ProtocolSerializer.kt` — register both hierarchies + +**acceptance criteria:** +- all messages serialize/deserialize correctly +- unknown message type produces `ProtocolError`, not crash +- protocol types are DTOs — no domain types leak into protocol layer + +--- + +## task 2: apps/server — Ktor server + websocket + +**dependencies:** task 1 + +**endpoints:** + +``` +GET /health → server + provider status +GET /sessions → list recent sessions (last 20) +GET /sessions/{id} → session detail + current state +POST /sessions → start new session (body: StartSessionRequest) +POST /sessions/{id}/cancel → cancel session +GET /sessions/{id}/events → full event log (paginated) +WS /sessions/{id}/stream → live event stream for session +WS /stream → global event stream (all sessions) +GET /workflows → list available workflow definitions +GET /providers → provider registry + health +``` + +**websocket behavior:** +- on connect: send snapshot of current session state, then stream live events +- on disconnect: clean up subscription, no state lost +- on reconnect: client sends last received event id, server replays missed events from that point +- heartbeat: server sends `Ping` every 30s, client must respond within 10s or connection dropped + +**files:** +- `apps/server/src/main/kotlin/.../Application.kt` +- `apps/server/src/main/kotlin/.../routes/SessionRoutes.kt` +- `apps/server/src/main/kotlin/.../routes/WorkflowRoutes.kt` +- `apps/server/src/main/kotlin/.../routes/ProviderRoutes.kt` +- `apps/server/src/main/kotlin/.../ws/SessionStreamHandler.kt` +- `apps/server/src/main/kotlin/.../ws/GlobalStreamHandler.kt` +- `apps/server/src/main/kotlin/.../ServerModule.kt` — wires InfrastructureModule + Ktor + +**acceptance criteria:** +- `/health` returns 200 with provider status +- websocket stream delivers events within 100ms of emission +- reconnect with last event id replays missed events correctly +- unknown WS message type → `ProtocolError` sent, connection kept open + +--- + +## task 3: apps/server — approval interaction + +**dependencies:** task 2 + +approval flow: +1. orchestrator emits `ApprovalRequiredEvent` → `OrchestrationPausedEvent` +2. server catches pause, sends `ApprovalRequired` message to all connected clients for that session +3. client sends `ApprovalResponse` +4. server calls `approvalEngine.grant()` or `approvalEngine.deny()` accordingly +5. orchestrator resumes + +**steering:** `ApprovalDecision.STEER` + `steeringNote` injects a steering event into the session event stream before resuming. the note becomes L0 context for the next stage. + +**timeout:** if no response within `ApprovalConfig.timeoutMs`, server auto-denies and emits `ApprovalTimedOutEvent`. configurable, default 5 minutes. + +**files:** +- `apps/server/src/main/kotlin/.../approval/ApprovalCoordinator.kt` +- `apps/server/src/main/kotlin/.../routes/ApprovalRoutes.kt` — REST fallback for non-WS clients + +**acceptance criteria:** +- approval required → session pauses → client receives message +- approve → session resumes +- reject → session fails with reason +- steer → steering note injected → session resumes with note in context +- timeout → auto-deny, event emitted + +--- + +## task 4: apps/tui — Mosaic dashboard + +**dependencies:** task 2, task 3 + +**layout:** +``` +┌─ status bar ──────────────────────────────────────────────────────┐ +│ server: ● connected │ provider: gemma-4 (loaded) │ gpu: 74% │ cwd │ +├─ session list ────────────────────────────────────────────────────┤ +│ ▶ [abc123] "fix auth bug" ACTIVE stage 3/5 2m ago │ +│ [def456] "refactor context" PAUSED awaiting approval │ +│ [ghi789] "add tool registry" DONE 5m ago │ +│ [jkl012] "write tests for..." FAILED 12m ago │ +│ [mno345] "epic 11 wiring" DONE 1h ago │ +├─ active session ──────────────────────────────────────────────────┤ +│ stage: execute_plan (3/5) │ +│ last output: "FileEditTool applied patch to src/main/..." │ +│ tool: FileEditTool ████████░░ T2 │ +├─ approval prompt (when pending) ──────────────────────────────────┤ +│ ⚠ APPROVAL REQUIRED — Tier T3 │ +│ tool: ShellTool argv: ["rm", "-rf", "build/"] │ +│ risk: HIGH — DestructiveOperation, RepeatedFailure(3/3) │ +│ [A] approve [R] reject [S] steer [D] details │ +├─ input ────────────────────────────────────────────────────────────┤ +│ > _ │ +├─ keybinds ────────────────────────────────────────────────────────┤ +│ n new r resume c cancel a approve q quit ? help , config │ +└───────────────────────────────────────────────────────────────────┘ +``` + +**keybinds:** +- `n` — new session (prompts for workflow selection) +- `r` — resume selected session +- `c` — cancel selected session +- `a/r/s` — approve/reject/steer (only active when approval pending) +- `↑↓` — navigate session list +- `enter` — select/expand session +- `q` — quit +- `,` — open config +- `?` — help overlay + +**state management:** Mosaic reactive state. server events update state via websocket listener coroutine. no polling. + +**files:** +- `apps/tui/src/main/kotlin/.../TuiApp.kt` — Mosaic entry point +- `apps/tui/src/main/kotlin/.../components/StatusBar.kt` +- `apps/tui/src/main/kotlin/.../components/SessionList.kt` +- `apps/tui/src/main/kotlin/.../components/ActiveSession.kt` +- `apps/tui/src/main/kotlin/.../components/ApprovalPrompt.kt` +- `apps/tui/src/main/kotlin/.../components/InputBar.kt` +- `apps/tui/src/main/kotlin/.../ws/TuiWsClient.kt` — websocket client +- `apps/tui/src/main/kotlin/.../state/TuiState.kt` + +**acceptance criteria:** +- status bar updates within 1s of provider state change +- session list shows last 5 sessions with correct status +- approval prompt appears immediately on `ApprovalRequired` message +- approve/reject/steer sends correct `ApprovalResponse` to server +- disconnect from server → status bar shows reconnecting state, no crash + +--- + +## task 5: apps/cli — Clikt entrypoint + +**dependencies:** task 2, task 3 + +**commands:** +``` +correx run --workflow [--session ] [--auto-approve] +correx session list [--limit 20] +correx session resume +correx session cancel +correx session events [--from ] +correx approve --decision [--note "..."] +correx status +correx provider list +correx config [--edit] +``` + +**output format:** +- default: human-readable, colored +- `--json` flag: machine-readable JSON on stdout for all commands +- `--quiet` flag: minimal output, only errors + +**`correx run` behavior:** +- starts session, subscribes to websocket stream +- prints stage transitions and tool calls to stdout +- on approval required: prompts interactively if TTY, auto-denies if `--auto-approve` not set and no TTY +- exits 0 on session complete, 1 on failure, 2 on approval denied + +**files:** +- `apps/cli/src/main/kotlin/.../CorrexCli.kt` — root Clikt command +- `apps/cli/src/main/kotlin/.../commands/RunCommand.kt` +- `apps/cli/src/main/kotlin/.../commands/SessionCommand.kt` +- `apps/cli/src/main/kotlin/.../commands/ApproveCommand.kt` +- `apps/cli/src/main/kotlin/.../commands/StatusCommand.kt` +- `apps/cli/src/main/kotlin/.../commands/ProviderCommand.kt` +- `apps/cli/src/main/kotlin/.../ws/CliWsClient.kt` + +**acceptance criteria:** +- `correx run` exits with correct codes +- `--json` produces valid parseable JSON for all commands +- approval prompt works in TTY, auto-denies without TTY when flag absent +- `correx session list` output matches server `/sessions` response + +--- + +## task 6: configuration + +**shared config file:** `~/.config/correx/config.toml` (or `$CORREX_CONFIG`) + +```toml +[server] +host = "localhost" +port = 8080 + +[tui] +theme = "dark" +session_list_limit = 5 + +[cli] +default_output = "human" + +[approval] +timeout_ms = 300000 # 5 minutes +``` + +config accessible from both TUI (`,` keybind) and CLI (`correx config --edit`). + +**files:** +- `apps/shared/src/main/kotlin/.../config/CorrexConfig.kt` +- `apps/shared/src/main/kotlin/.../config/ConfigLoader.kt` + +--- + +## sequencing + +``` +task 1 (protocol) + ↓ +task 2 (server + websocket) + ↓ +task 3 (approval interaction) ← depends on task 2 + ↓ +task 4 (TUI) task 5 (CLI) ← both depend on tasks 2+3, parallel + ↓ +task 6 (config) ← shared, can be done alongside 4+5 +``` + +--- + +## deferred to epic 14+ + +- router layer (L2/L3 memory bridge) +- workflow definition editor in TUI +- session diff viewer +- replay viewer (epic 15) +- GPU memory pressure indicator beyond simple % usage +- multi-server TUI (connecting to remote correx instances) diff --git a/infrastructure/inference/src/main/kotlin/com/correx/infrastructure/inference/FirstAvailableRoutingStrategy.kt b/infrastructure/inference/src/main/kotlin/com/correx/infrastructure/inference/FirstAvailableRoutingStrategy.kt index 4cffb7e9..572de78c 100644 --- a/infrastructure/inference/src/main/kotlin/com/correx/infrastructure/inference/FirstAvailableRoutingStrategy.kt +++ b/infrastructure/inference/src/main/kotlin/com/correx/infrastructure/inference/FirstAvailableRoutingStrategy.kt @@ -10,7 +10,7 @@ import com.correx.core.inference.RoutingStrategy * Selects the first [InferenceProvider] from [candidates] whose declared capabilities * cover all [requiredCapabilities]. * - * "First" means the list order as supplied by [ProviderRegistry.resolve] — callers that + * "First" means the list order as supplied by [com.correx.core.inference.ProviderRegistry.resolve] — callers that * want score-ordered selection should pass a score-sorted list. * * @throws NoEligibleProviderException if no candidate satisfies all required capabilities. diff --git a/settings.gradle b/settings.gradle index 108f2707..f8ce649f 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,7 +1,15 @@ +pluginManagement { + repositories { + mavenCentral() + gradlePluginPortal() + } +} + rootProject.name = 'correx' include ':apps:cli' include ':apps:server' +include ':apps:tui' include ':apps:worker' include ':apps:desktop' diff --git a/testing/deterministic/src/test/kotlin/ValidationPipelineShortCircuitTest.kt b/testing/deterministic/src/test/kotlin/ValidationPipelineShortCircuitTest.kt index a5b4dcaa..48ecabaa 100644 --- a/testing/deterministic/src/test/kotlin/ValidationPipelineShortCircuitTest.kt +++ b/testing/deterministic/src/test/kotlin/ValidationPipelineShortCircuitTest.kt @@ -14,7 +14,7 @@ class ValidationPipelineShortCircuitTest { private val context = ValidationContext(WorkflowFixtures.simpleGraph()) - private fun errorValidator(name: String) = Validator { _ -> + private fun errorValidator(name: String = "first") = Validator { _ -> ValidationSection( name = name, issues = listOf(ValidationIssue("ERR", "failure", ValidationSeverity.ERROR)) @@ -29,7 +29,7 @@ class ValidationPipelineShortCircuitTest { ValidationSection(name = "second") } - val pipeline = ValidationPipeline(listOf(errorValidator("first"), trackingValidator)) + val pipeline = ValidationPipeline(listOf(errorValidator(), trackingValidator)) val outcome = pipeline.validate(context) as ValidationOutcome.Rejected assertFalse(secondCalled, "second validator must not run after a rejection") diff --git a/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt b/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt index bc73ed49..f06d8c50 100644 --- a/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt +++ b/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt @@ -1,9 +1,11 @@ package com.correx.testing.fixtures import com.correx.core.context.model.ContextPack +import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.NewEvent 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 @@ -126,9 +128,9 @@ object InferenceFixtures { tokensUsed: TokenUsage, latencyMs: Long, ): NewEvent { - return com.correx.core.events.events.NewEvent( - metadata = com.correx.core.events.events.EventMetadata( - eventId = com.correx.core.events.types.EventId("inf-$requestId"), + return NewEvent( + metadata = EventMetadata( + eventId = EventId("inf-$requestId"), sessionId = sessionId, timestamp = Clock.System.now(), schemaVersion = 1, diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index ea243e8b..8a995130 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -66,7 +66,7 @@ class SessionOrchestratorIntegrationTest { val orchestrationRepository = OrchestrationRepository(orchestrationReplayer) private val inferenceRepository = InferenceRepository( object : EventReplayer { - override fun rebuild(sessionId: com.correx.core.events.types.SessionId) = InferenceState() + override fun rebuild(sessionId: SessionId) = InferenceState() }, ) private val riskAssessor = DefaultRiskAssessor() @@ -270,7 +270,7 @@ class SessionOrchestratorIntegrationTest { val sessionId = SessionId("s7") val ex = org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException::class.java) { - kotlinx.coroutines.runBlocking { orchestrator.run(sessionId, brokenGraph, config) } + runBlocking { orchestrator.run(sessionId, brokenGraph, config) } } assertTrue(ex.message?.contains("ghost") == true, "Expected message to mention 'ghost', got: ${ex.message}") } diff --git a/testing/integration/src/test/kotlin/ValidationPipelineIntegrationTest.kt b/testing/integration/src/test/kotlin/ValidationPipelineIntegrationTest.kt index 9154829d..79b291f7 100644 --- a/testing/integration/src/test/kotlin/ValidationPipelineIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/ValidationPipelineIntegrationTest.kt @@ -52,7 +52,7 @@ class ValidationPipelineIntegrationTest { } @Test - fun `rejected outcome retryable is false — set by validator, not orchestrator`() { + fun `rejected outcome retryable is false - set by validator, not orchestrator`() { // A dangling transition triggers GraphValidator ERROR → Rejected(retryable=false). // Ownership rule: the validator sets retryable; the orchestrator must only read it. val a = StageId("A")