epic-13: add cli and tui entry point, finish epic.
This commit is contained in:
@@ -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"
|
||||
}
|
||||
|
||||
@@ -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<Throwable> { 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)
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
)
|
||||
@@ -0,0 +1,5 @@
|
||||
package com.correx.apps.server.approval
|
||||
|
||||
data class ApprovalConfig(
|
||||
val timeoutMs: Long = 300_000L,
|
||||
)
|
||||
@@ -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<SessionId, MutableSet<DefaultWebSocketServerSession>> =
|
||||
ConcurrentHashMap()
|
||||
private val resolved: ConcurrentHashMap<ApprovalRequestId, Boolean> = ConcurrentHashMap()
|
||||
private val timeoutJobs: ConcurrentHashMap<ApprovalRequestId, Job> = 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,
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.correx.apps.server.protocol
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
@Serializable
|
||||
data class RiskSummaryDto(
|
||||
val level: String,
|
||||
val factors: List<String>,
|
||||
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,
|
||||
}
|
||||
@@ -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<ClientMessage>(json)
|
||||
}.getOrElse { cause ->
|
||||
throw ProtocolException("Failed to decode client message", cause)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
@@ -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<InferenceProvider>
|
||||
suspend fun healthCheckAll(): Map<ProviderId, ProviderHealth>
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package com.correx.apps.server.registry
|
||||
|
||||
import com.correx.core.transitions.graph.WorkflowGraph
|
||||
|
||||
interface WorkflowRegistry {
|
||||
fun listAll(): List<WorkflowSummary>
|
||||
fun find(workflowId: String): WorkflowGraph?
|
||||
}
|
||||
|
||||
data class WorkflowSummary(
|
||||
val workflowId: String,
|
||||
val description: String,
|
||||
)
|
||||
@@ -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<ApprovalRequest>()
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
@@ -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<SessionSummaryResponse>()
|
||||
call.respond(sessions)
|
||||
}
|
||||
|
||||
post {
|
||||
val body = call.receive<StartSessionRequest>()
|
||||
module.workflowRegistry.find(body.workflowId)
|
||||
?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: ${body.workflowId}")
|
||||
|
||||
val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
|
||||
call.respond(HttpStatusCode.Created, StartSessionResponse(sessionId.value))
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<WorkflowSummary> = module.workflowRegistry.listAll()
|
||||
call.respond(workflows)
|
||||
}
|
||||
}
|
||||
@@ -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)))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user