From 7d46f46f017b2ca6ff3988cd70f38affe0339309 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 17 May 2026 03:17:12 +0400 Subject: [PATCH] fix(server): event capturing and necessary logic for smoke test. --- apps/server/build.gradle | 12 ++ .../kotlin/com/correx/apps/server/Main.kt | 98 ++++++++++- .../registry/FileSystemWorkflowRegistry.kt | 39 +++++ .../apps/server/routes/SessionRoutes.kt | 146 ++++++++++------ .../apps/server/ws/GlobalStreamHandler.kt | 86 +++++++++- .../repository/ArtifactRepository.kt | 13 ++ .../orchestration/OrchestratorEngines.kt | 2 + .../orchestration/OrchestratorRepositories.kt | 2 + core/transitions/build.gradle | 1 + .../core/transitions/conditions/AlwaysTrue.kt | 8 + .../conditions/ArtifactConditions.kt | 21 +++ .../core/transitions/conditions/Composites.kt | 19 +++ .../transitions/conditions/VariableEquals.kt | 9 + .../evaluation/EvaluationContext.kt | 5 +- .../transitions/evaluation/PromptResolver.kt | 5 + .../TransitionConditionEvaluator.kt | 2 +- .../core/transitions/graph/StageConfig.kt | 4 + .../core/transitions/graph/WorkflowGraph.kt | 1 + .../conditions/TransitionConditionTest.kt | 159 ++++++++++++++++++ infrastructure/build.gradle | 2 + infrastructure/persistence/build.gradle | 1 + .../artifact/LiveArtifactRepository.kt | 96 +++++++++++ .../infrastructure/InfrastructureModule.kt | 45 ++++- infrastructure/workflow/build.gradle | 14 ++ .../workflow/ConditionFactory.kt | 39 +++++ .../infrastructure/workflow/ConditionSpec.kt | 10 ++ .../workflow/FileSystemPromptLoader.kt | 19 +++ .../infrastructure/workflow/PromptLoader.kt | 5 + .../workflow/PromptNotFoundException.kt | 3 + .../workflow/TomlWorkflowLoader.kt | 130 ++++++++++++++ .../infrastructure/workflow/WorkflowLoader.kt | 8 + .../workflow/WorkflowValidationException.kt | 3 + .../workflow/FileSystemPromptLoaderTest.kt | 45 +++++ .../workflow/TomlWorkflowLoaderTest.kt | 93 ++++++++++ prompts/healthcheck_execute.md | 4 + prompts/healthcheck_write.md | 10 ++ settings.gradle | 1 + testing/integration/build.gradle | 1 + .../SessionOrchestratorIntegrationTest.kt | 3 + 39 files changed, 1097 insertions(+), 67 deletions(-) create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt create mode 100644 core/artifacts/src/main/kotlin/com/correx/core/artifacts/repository/ArtifactRepository.kt create mode 100644 core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/AlwaysTrue.kt create mode 100644 core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/ArtifactConditions.kt create mode 100644 core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/Composites.kt create mode 100644 core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/VariableEquals.kt create mode 100644 core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/PromptResolver.kt create mode 100644 core/transitions/src/test/kotlin/com/correx/core/transitions/conditions/TransitionConditionTest.kt create mode 100644 infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt create mode 100644 infrastructure/workflow/build.gradle create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionFactory.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionSpec.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoader.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptLoader.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptNotFoundException.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowLoader.kt create mode 100644 infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowValidationException.kt create mode 100644 infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoaderTest.kt create mode 100644 infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoaderTest.kt create mode 100644 prompts/healthcheck_execute.md create mode 100644 prompts/healthcheck_write.md diff --git a/apps/server/build.gradle b/apps/server/build.gradle index 26b23e1d..0164233b 100644 --- a/apps/server/build.gradle +++ b/apps/server/build.gradle @@ -15,9 +15,21 @@ dependencies { implementation project(':core:kernel') implementation project(':core:inference') implementation project(':core:transitions') + implementation project(':core:context') + implementation project(':core:validation') + implementation project(':core:risk') + implementation project(':core:artifacts') + implementation project(':infrastructure') + implementation project(':infrastructure:workflow') + implementation project(':infrastructure:persistence') + implementation project(':infrastructure:inference') + implementation project(':infrastructure:inference:llama_cpp') + implementation project(':infrastructure:inference:commons') implementation "com.github.ajalt.clikt:clikt:5.0.1" + implementation "io.ktor:ktor-client-cio:$ktor_version" + implementation "io.ktor:ktor-client-logging:$ktor_version" implementation "io.ktor:ktor-server-core:$ktor_version" implementation "io.ktor:ktor-server-netty:$ktor_version" implementation "io.ktor:ktor-server-websockets:$ktor_version" diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 80f574a4..7bd14349 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -1,5 +1,101 @@ package com.correx.apps.server +import com.correx.apps.server.registry.FileSystemWorkflowRegistry +import com.correx.apps.server.registry.ProviderRegistry +import com.correx.core.approvals.domain.DefaultApprovalEngine +import com.correx.core.context.builder.DefaultContextPackBuilder +import com.correx.core.context.compression.DefaultContextCompressor +import com.correx.core.inference.DefaultInferenceRouter +import com.correx.core.inference.InferenceProjector +import com.correx.core.inference.InferenceRepository +import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer +import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.kernel.orchestration.OrchestrationProjector +import com.correx.core.kernel.orchestration.OrchestrationRepository +import com.correx.core.kernel.orchestration.OrchestratorEngines +import com.correx.core.kernel.orchestration.OrchestratorRepositories +import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.risk.DefaultRiskAssessor +import com.correx.core.sessions.DefaultSessionReducer +import com.correx.core.sessions.DefaultSessionRepository +import com.correx.core.sessions.SessionProjector +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.transitions.evaluation.PromptResolver +import com.correx.core.transitions.resolution.DefaultTransitionResolver +import com.correx.core.validation.pipeline.ValidationPipeline +import com.correx.infrastructure.InfrastructureModule +import com.correx.infrastructure.inference.DefaultProviderRegistry +import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy +import io.ktor.server.engine.embeddedServer +import io.ktor.server.netty.Netty + fun main() { - println("correx :: apps/server") + val eventStore = InfrastructureModule.createEventStore() + + val llamaProvider = InfrastructureModule.createLlamaCppProvider( + modelId = System.getenv("CORREX_MODEL_ID") ?: "default", + modelPath = System.getenv("CORREX_MODEL_PATH") ?: "", + baseUrl = System.getenv("CORREX_LLAMA_URL") ?: "http://127.0.0.1:10000", + ) + val infraRegistry = InfrastructureModule.createProviderRegistry(listOf(llamaProvider)) + + val artifactRepository = InfrastructureModule.createArtifactRepository(eventStore) + + val sessionRepository = DefaultSessionRepository( + DefaultEventReplayer(eventStore, SessionProjector(DefaultSessionReducer())) + ) + val inferenceRepository = InferenceRepository( + DefaultEventReplayer(eventStore, InferenceProjector()) + ) + val orchestrationRepository = OrchestrationRepository( + DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())) + ) + + val inferenceRouter = DefaultInferenceRouter(infraRegistry, FirstAvailableRoutingStrategy()) + + val promptLoader = InfrastructureModule.createPromptLoader() + val promptResolver = PromptResolver { path -> promptLoader.load(path) } + + val repositories = OrchestratorRepositories( + eventStore = eventStore, + inferenceRepository = inferenceRepository, + orchestrationRepository = orchestrationRepository, + sessionRepository = sessionRepository, + artifactRepository = artifactRepository, + ) + + val engines = OrchestratorEngines( + transitionResolver = DefaultTransitionResolver { condition, ctx -> condition.evaluate(ctx) }, + contextPackBuilder = DefaultContextPackBuilder(DefaultContextCompressor()), + inferenceRouter = inferenceRouter, + validationPipeline = ValidationPipeline(validators = emptyList()), + approvalEngine = DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + promptResolver = promptResolver, + ) + + val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + ) + + val workflowRegistry = FileSystemWorkflowRegistry(InfrastructureModule.createWorkflowLoader()) + + val module = ServerModule( + orchestrator = orchestrator, + eventStore = eventStore, + sessionRepository = sessionRepository, + workflowRegistry = workflowRegistry, + providerRegistry = infraRegistry.asServerRegistry(), + ) + + embeddedServer(Netty, port = 8080) { + configureServer(module) + }.start(wait = true) +} + +private fun DefaultProviderRegistry.asServerRegistry(): ProviderRegistry = object : ProviderRegistry { + override fun listAll() = this@asServerRegistry.listAll() + override suspend fun healthCheckAll() = this@asServerRegistry.healthCheckAll() } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt b/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt new file mode 100644 index 00000000..6f953dcd --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt @@ -0,0 +1,39 @@ +package com.correx.apps.server.registry + +import com.correx.core.transitions.graph.WorkflowGraph +import com.correx.infrastructure.workflow.WorkflowLoader +import java.nio.file.Path +import kotlin.io.path.exists +import kotlin.io.path.listDirectoryEntries + +class FileSystemWorkflowRegistry( + private val loader: WorkflowLoader, + private val workflowsDir: Path = Path.of( + System.getProperty("user.home"), ".config", "correx", "workflows" + ), +) : WorkflowRegistry { + + private val cache: Map by lazy { loadAll() } + + override fun listAll(): List = + cache.values.map { WorkflowSummary(workflowId = it.id, description = it.id) } + + override fun find(workflowId: String): WorkflowGraph? = cache[workflowId] + + private fun loadAll(): Map { + if (!workflowsDir.exists()) return emptyMap() + return workflowsDir.listDirectoryEntries("*.toml") + .mapNotNull { file -> + runCatching { loader.load(file) } + .onFailure { logWarning(file, it) } + .getOrNull() + } + .associateBy { graph -> + graph.id.ifBlank { graph.let { it } .let { file -> file.start.value } } + } + } + + private fun logWarning(file: Path, ex: Throwable) { + System.err.println("[WorkflowRegistry] Skipping $file: ${ex.message}") + } +} 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 index ac01283c..af841d88 100644 --- 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 @@ -3,10 +3,16 @@ package com.correx.apps.server.routes import com.correx.apps.server.ServerModule import com.correx.apps.server.protocol.SessionConfigDto import com.correx.apps.server.ws.SessionStreamHandler +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId +import com.correx.core.kernel.orchestration.OrchestrationConfig import com.correx.core.utils.TypeId import io.ktor.http.HttpStatusCode import io.ktor.server.application.call +import io.ktor.server.application.application import io.ktor.server.request.receive import io.ktor.server.response.respond import io.ktor.server.routing.Route @@ -14,6 +20,8 @@ import io.ktor.server.routing.get import io.ktor.server.routing.post import io.ktor.server.routing.route import io.ktor.server.websocket.webSocket +import kotlinx.coroutines.launch +import kotlinx.datetime.Clock import kotlinx.serialization.Serializable import java.util.UUID @@ -34,67 +42,95 @@ data class StartSessionResponse(val sessionId: String) fun Route.sessionRoutes(module: ServerModule) { val streamHandler = SessionStreamHandler(module) - route("/sessions") { - get { - val sessions = emptyList() - 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)) - } - + get { call.respond(emptyList()) } + startSessionRoute(module) route("/{id}") { - get { - val id = call.parameters["id"] - ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id") - val sessionId: SessionId = TypeId(id) - val session = runCatching { module.sessionRepository.getSession(sessionId) }.getOrNull() - ?: return@get call.respond(HttpStatusCode.NotFound, "Session not found") - val response = SessionStateResponse( - sessionId = session.sessionId.value, - status = session.state.status.name, - createdAt = session.state.createdAt?.toString(), - ) - call.respond(response) - } - - post("/cancel") { - val id = call.parameters["id"] - ?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id") - val sessionId: SessionId = TypeId(id) - module.orchestrator.cancel(sessionId) - call.respond(HttpStatusCode.OK) - } - - get("/events") { - val id = call.parameters["id"] - ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id") - val from = call.request.queryParameters["from"]?.toLongOrNull() ?: 0L - val sessionId: SessionId = TypeId(id) - val events = module.eventStore.readFrom(sessionId, from) - val response = events.map { e -> - EventResponse( - eventId = e.metadata.eventId.value, - sequence = e.sequence, - sessionId = e.metadata.sessionId.value, - ) - } - call.respond(response) - } - + getSessionRoute(module) + cancelSessionRoute(module) + getEventsRoute(module) webSocket("/stream") { val id = call.parameters["id"] ?: return@webSocket val lastEventId = call.request.queryParameters["lastEventId"]?.toLongOrNull() - val sessionId: SessionId = TypeId(id) - streamHandler.handle(this, sessionId, lastEventId) + streamHandler.handle(this, TypeId(id), lastEventId) } } } } + +private fun Route.startSessionRoute(module: ServerModule) { + post { + val body = call.receive() + val graph = module.workflowRegistry.find(body.workflowId) + ?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: ${body.workflowId}") + val sessionId: SessionId = TypeId(UUID.randomUUID().toString()) + call.application.launch { + runCatching { module.orchestrator.run(sessionId, graph, OrchestrationConfig()) } + .onFailure { ex -> + System.err.println("[SessionRoutes] run failed for $sessionId: ${ex.message}") + module.eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = WorkflowFailedEvent( + sessionId = sessionId, + stageId = graph.start, + reason = ex.message ?: "unexpected orchestrator failure", + retryExhausted = false, + ), + ) + ) + } + } + call.respond(HttpStatusCode.Accepted, StartSessionResponse(sessionId.value)) + } +} + +private fun Route.getSessionRoute(module: ServerModule) { + get { + val id = call.parameters["id"] + ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id") + val session = runCatching { module.sessionRepository.getSession(TypeId(id)) }.getOrNull() + ?: return@get call.respond(HttpStatusCode.NotFound, "Session not found") + call.respond( + SessionStateResponse( + sessionId = session.sessionId.value, + status = session.state.status.name, + createdAt = session.state.createdAt?.toString(), + ) + ) + } +} + +private fun Route.cancelSessionRoute(module: ServerModule) { + post("/cancel") { + val id = call.parameters["id"] + ?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id") + module.orchestrator.cancel(TypeId(id)) + call.respond(HttpStatusCode.OK) + } +} + +private fun Route.getEventsRoute(module: ServerModule) { + get("/events") { + val id = call.parameters["id"] + ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id") + val from = call.request.queryParameters["from"]?.toLongOrNull() ?: 0L + val events = module.eventStore.readFrom(TypeId(id), from) + call.respond( + events.map { e -> + EventResponse( + eventId = e.metadata.eventId.value, + sequence = e.sequence, + sessionId = e.metadata.sessionId.value, + ) + } + ) + } +} 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 index d2393b32..8219b51a 100644 --- 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 @@ -3,7 +3,16 @@ package com.correx.apps.server.ws import com.correx.apps.server.ServerModule import com.correx.apps.server.protocol.ClientMessage import com.correx.apps.server.protocol.ProtocolSerializer +import com.correx.apps.server.protocol.ProviderHealthDto import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.inference.ProviderHealth +import com.correx.core.kernel.orchestration.OrchestrationConfig +import com.correx.core.utils.TypeId import io.ktor.server.websocket.DefaultWebSocketServerSession import io.ktor.websocket.Frame import io.ktor.websocket.readText @@ -11,12 +20,15 @@ import kotlinx.coroutines.channels.ClosedReceiveChannelException import kotlinx.coroutines.delay import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.datetime.Clock +import java.util.* private const val HEARTBEAT_INTERVAL_MS = 30_000L class GlobalStreamHandler(private val module: ServerModule) { suspend fun handle(session: DefaultWebSocketServerSession) { + System.err.println("[GlobalStream] client connected") sendInitialSnapshot(session) val heartbeatJob = session.launch { @@ -32,26 +44,39 @@ class GlobalStreamHandler(private val module: ServerModule) { if (frame is Frame.Text) { val text = frame.readText() runCatching { ProtocolSerializer.decodeClientMessage(text) } - .onSuccess { msg -> handleClientMessage(session, msg) } + .onSuccess { msg -> + System.err.println("[GlobalStream] recv: ${msg::class.simpleName}") + handleClientMessage(session, msg) + } .onFailure { + System.err.println("[GlobalStream] decode error: ${it.message}") val error = ServerMessage.ProtocolError("Unknown message: ${it.message}") session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) } } } } catch (_: ClosedReceiveChannelException) { - // client disconnected + System.err.println("[GlobalStream] client disconnected") } finally { heartbeatJob.cancel() } } private suspend fun sendInitialSnapshot(session: DefaultWebSocketServerSession) { - val providerHealth = runCatching { module.providerRegistry.healthCheckAll() }.getOrDefault(emptyMap()) - providerHealth.forEach { (providerId, _) -> + val providerHealth = runCatching { module.providerRegistry.healthCheckAll() } + .onFailure { System.err.println("[GlobalStream] healthCheckAll failed: ${it.message}") } + .getOrDefault(emptyMap()) + System.err.println("[GlobalStream] snapshot: ${providerHealth.size} provider(s)") + providerHealth.forEach { (providerId, health) -> + val status = when (health) { + is ProviderHealth.Healthy -> "healthy" + is ProviderHealth.Degraded -> "degraded" + is ProviderHealth.Unavailable -> "unavailable" + } + System.err.println("[GlobalStream] provider=${providerId.value} status=$status") val msg = ServerMessage.ProviderStatusChanged( providerId = providerId.value, - status = com.correx.apps.server.protocol.ProviderHealthDto(providerId.value, "unknown", null), + status = ProviderHealthDto(providerId.value, status, null), ) session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg))) } @@ -60,8 +85,55 @@ class GlobalStreamHandler(private val module: ServerModule) { private suspend fun handleClientMessage(session: DefaultWebSocketServerSession, msg: ClientMessage) { when (msg) { is ClientMessage.Ping -> Unit - else -> { - val error = ServerMessage.ProtocolError("Unexpected message type in global stream") + + is ClientMessage.StartSession -> { + val graph = module.workflowRegistry.find(msg.workflowId) + if (graph == null) { + val error = ServerMessage.ProtocolError("Unknown workflowId: ${msg.workflowId}") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) + return + } + val sessionId: SessionId = TypeId(UUID.randomUUID().toString()) + System.err.println("[GlobalStream] starting session=${sessionId.value} workflow=${msg.workflowId}") + session.launch { + runCatching { module.orchestrator.run(sessionId, graph, OrchestrationConfig()) } + .onFailure { ex -> + System.err.println("[GlobalStreamHandler] run failed for $sessionId: ${ex.message}") + module.eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = WorkflowFailedEvent( + sessionId = sessionId, + stageId = graph.start, + reason = ex.message ?: "unexpected orchestrator failure", + retryExhausted = false, + ), + ), + ) + } + } + val started = ServerMessage.SessionStarted(sessionId = sessionId, workflowId = msg.workflowId) + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) + } + + is ClientMessage.CancelSession -> { + module.orchestrator.cancel(msg.sessionId) + } + + is ClientMessage.ResumeSession -> { + val error = ServerMessage.ProtocolError("ResumeSession not supported") + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) + } + + is ClientMessage.ApprovalResponse -> { + val error = ServerMessage.ProtocolError("ApprovalResponse not supported on global stream") session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) } } diff --git a/core/artifacts/src/main/kotlin/com/correx/core/artifacts/repository/ArtifactRepository.kt b/core/artifacts/src/main/kotlin/com/correx/core/artifacts/repository/ArtifactRepository.kt new file mode 100644 index 00000000..4ea52afe --- /dev/null +++ b/core/artifacts/src/main/kotlin/com/correx/core/artifacts/repository/ArtifactRepository.kt @@ -0,0 +1,13 @@ +package com.correx.core.artifacts.repository + +import com.correx.core.artifacts.ArtifactState +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId + +interface ArtifactRepository { + fun getBySession(sessionId: SessionId): Map + fun getByStage(sessionId: SessionId, stageId: StageId): Map + fun getById(sessionId: SessionId, artifactId: ArtifactId): ArtifactState? + fun getByIds(sessionId: SessionId, artifactIds: Set): Map +} diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt index a36bd2ee..85e0bd93 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorEngines.kt @@ -4,6 +4,7 @@ import com.correx.core.approvals.domain.ApprovalEngine import com.correx.core.context.builder.ContextPackBuilder import com.correx.core.inference.InferenceRouter import com.correx.core.risk.RiskAssessor +import com.correx.core.transitions.evaluation.PromptResolver import com.correx.core.transitions.resolution.TransitionResolver import com.correx.core.validation.pipeline.ValidationPipeline @@ -14,4 +15,5 @@ data class OrchestratorEngines( val validationPipeline: ValidationPipeline, val approvalEngine: ApprovalEngine, val riskAssessor: RiskAssessor, + val promptResolver: PromptResolver = PromptResolver { "" }, ) diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt index 45fba516..56d4c8fc 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/OrchestratorRepositories.kt @@ -1,5 +1,6 @@ package com.correx.core.kernel.orchestration +import com.correx.core.artifacts.repository.ArtifactRepository import com.correx.core.events.stores.EventStore import com.correx.core.inference.InferenceRepository import com.correx.core.sessions.DefaultSessionRepository @@ -9,4 +10,5 @@ data class OrchestratorRepositories( val inferenceRepository: InferenceRepository, val orchestrationRepository: OrchestrationRepository, val sessionRepository: DefaultSessionRepository, + val artifactRepository: ArtifactRepository, ) diff --git a/core/transitions/build.gradle b/core/transitions/build.gradle index 541e4ca1..c348ef3d 100644 --- a/core/transitions/build.gradle +++ b/core/transitions/build.gradle @@ -7,4 +7,5 @@ plugins { dependencies { implementation(project(":core:events")) implementation(project(":core:inference")) + implementation(project(":core:artifacts")) } \ No newline at end of file diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/AlwaysTrue.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/AlwaysTrue.kt new file mode 100644 index 00000000..2beff0bd --- /dev/null +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/AlwaysTrue.kt @@ -0,0 +1,8 @@ +package com.correx.core.transitions.conditions + +import com.correx.core.transitions.evaluation.EvaluationContext +import com.correx.core.transitions.graph.TransitionCondition + +object AlwaysTrue : TransitionCondition { + override fun evaluate(context: EvaluationContext) = true +} diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/ArtifactConditions.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/ArtifactConditions.kt new file mode 100644 index 00000000..a008f2f9 --- /dev/null +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/ArtifactConditions.kt @@ -0,0 +1,21 @@ +package com.correx.core.transitions.conditions + +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ArtifactLifecyclePhase +import com.correx.core.transitions.evaluation.EvaluationContext +import com.correx.core.transitions.graph.TransitionCondition + +data class ArtifactPresent(val artifactId: ArtifactId) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + context.artifacts[artifactId] != null +} + +data class ArtifactAbsent(val artifactId: ArtifactId) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + context.artifacts[artifactId] == null +} + +data class ArtifactValidated(val artifactId: ArtifactId) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + context.artifacts[artifactId]?.phase == ArtifactLifecyclePhase.VALIDATED +} diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/Composites.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/Composites.kt new file mode 100644 index 00000000..36e4d57f --- /dev/null +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/Composites.kt @@ -0,0 +1,19 @@ +package com.correx.core.transitions.conditions + +import com.correx.core.transitions.evaluation.EvaluationContext +import com.correx.core.transitions.graph.TransitionCondition + +data class AllOf(val conditions: List) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + conditions.all { it.evaluate(context) } +} + +data class AnyOf(val conditions: List) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + conditions.any { it.evaluate(context) } +} + +data class Not(val condition: TransitionCondition) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + !condition.evaluate(context) +} diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/VariableEquals.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/VariableEquals.kt new file mode 100644 index 00000000..e4b660ba --- /dev/null +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/conditions/VariableEquals.kt @@ -0,0 +1,9 @@ +package com.correx.core.transitions.conditions + +import com.correx.core.transitions.evaluation.EvaluationContext +import com.correx.core.transitions.graph.TransitionCondition + +data class VariableEquals(val key: String, val value: String) : TransitionCondition { + override fun evaluate(context: EvaluationContext) = + context.variables[key] == value +} diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/EvaluationContext.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/EvaluationContext.kt index b078d48e..137e7c4b 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/EvaluationContext.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/EvaluationContext.kt @@ -1,10 +1,13 @@ package com.correx.core.transitions.evaluation +import com.correx.core.artifacts.ArtifactState +import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId data class EvaluationContext( val sessionId: SessionId, val currentStage: StageId, - val variables: Map + val artifacts: Map = emptyMap(), + val variables: Map = emptyMap(), ) \ No newline at end of file diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/PromptResolver.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/PromptResolver.kt new file mode 100644 index 00000000..5b59c83c --- /dev/null +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/PromptResolver.kt @@ -0,0 +1,5 @@ +package com.correx.core.transitions.evaluation + +fun interface PromptResolver { + fun resolve(path: String): String +} diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/TransitionConditionEvaluator.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/TransitionConditionEvaluator.kt index 3085576a..99496c72 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/TransitionConditionEvaluator.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/evaluation/TransitionConditionEvaluator.kt @@ -2,7 +2,7 @@ package com.correx.core.transitions.evaluation import com.correx.core.transitions.graph.TransitionCondition -interface TransitionConditionEvaluator { +fun interface TransitionConditionEvaluator { fun evaluate( condition: TransitionCondition, context: EvaluationContext diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt index 657bd79d..7f6827bb 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/StageConfig.kt @@ -1,5 +1,6 @@ package com.correx.core.transitions.graph +import com.correx.core.events.types.ArtifactId import com.correx.core.inference.GenerationConfig import com.correx.core.inference.ModelCapability import kotlinx.serialization.Serializable @@ -15,4 +16,7 @@ data class StageConfig( ), val allowedTools: Set = emptySet(), val maxRetries: Int = 3, + val produces: Set = emptySet(), + val needs: Set = emptySet(), + val metadata: Map = emptyMap(), ) diff --git a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/WorkflowGraph.kt b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/WorkflowGraph.kt index ad3c671f..0838eb24 100644 --- a/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/WorkflowGraph.kt +++ b/core/transitions/src/main/kotlin/com/correx/core/transitions/graph/WorkflowGraph.kt @@ -3,6 +3,7 @@ package com.correx.core.transitions.graph import com.correx.core.events.types.StageId data class WorkflowGraph( + val id: String = "", val stages: Map, val transitions: Set, val start: StageId, diff --git a/core/transitions/src/test/kotlin/com/correx/core/transitions/conditions/TransitionConditionTest.kt b/core/transitions/src/test/kotlin/com/correx/core/transitions/conditions/TransitionConditionTest.kt new file mode 100644 index 00000000..3908f9bd --- /dev/null +++ b/core/transitions/src/test/kotlin/com/correx/core/transitions/conditions/TransitionConditionTest.kt @@ -0,0 +1,159 @@ +package com.correx.core.transitions.conditions + +import com.correx.core.artifacts.ArtifactState +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ArtifactLifecyclePhase +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.transitions.evaluation.EvaluationContext +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class TransitionConditionTest { + + private val baseContext = EvaluationContext( + sessionId = SessionId("test-session"), + currentStage = StageId("stage-1"), + ) + + private fun contextWith( + artifacts: Map = emptyMap(), + variables: Map = emptyMap(), + ) = baseContext.copy(artifacts = artifacts, variables = variables) + + // AlwaysTrue + + @Test + fun `AlwaysTrue returns true`() { + assertTrue(AlwaysTrue.evaluate(baseContext)) + } + + // ArtifactPresent + + @Test + fun `ArtifactPresent returns true when artifact in map`() { + val id = ArtifactId("a1") + val ctx = contextWith(artifacts = mapOf(id to ArtifactState())) + assertTrue(ArtifactPresent(id).evaluate(ctx)) + } + + @Test + fun `ArtifactPresent returns false when artifact absent`() { + val ctx = contextWith() + assertFalse(ArtifactPresent(ArtifactId("a1")).evaluate(ctx)) + } + + // ArtifactAbsent + + @Test + fun `ArtifactAbsent returns false when artifact in map`() { + val id = ArtifactId("a1") + val ctx = contextWith(artifacts = mapOf(id to ArtifactState())) + assertFalse(ArtifactAbsent(id).evaluate(ctx)) + } + + @Test + fun `ArtifactAbsent returns true when artifact absent`() { + val ctx = contextWith() + assertTrue(ArtifactAbsent(ArtifactId("a1")).evaluate(ctx)) + } + + // ArtifactValidated + + @Test + fun `ArtifactValidated returns true when phase is VALIDATED`() { + val id = ArtifactId("a1") + val ctx = contextWith(artifacts = mapOf(id to ArtifactState(phase = ArtifactLifecyclePhase.VALIDATED))) + assertTrue(ArtifactValidated(id).evaluate(ctx)) + } + + @Test + fun `ArtifactValidated returns false when phase is CREATED`() { + val id = ArtifactId("a1") + val ctx = contextWith(artifacts = mapOf(id to ArtifactState(phase = ArtifactLifecyclePhase.CREATED))) + assertFalse(ArtifactValidated(id).evaluate(ctx)) + } + + @Test + fun `ArtifactValidated returns false when phase is REJECTED`() { + val id = ArtifactId("a1") + val ctx = contextWith(artifacts = mapOf(id to ArtifactState(phase = ArtifactLifecyclePhase.REJECTED))) + assertFalse(ArtifactValidated(id).evaluate(ctx)) + } + + @Test + fun `ArtifactValidated returns false when artifact absent`() { + val ctx = contextWith() + assertFalse(ArtifactValidated(ArtifactId("a1")).evaluate(ctx)) + } + + // VariableEquals + + @Test + fun `VariableEquals returns true when key matches value`() { + val ctx = contextWith(variables = mapOf("status" to "done")) + assertTrue(VariableEquals("status", "done").evaluate(ctx)) + } + + @Test + fun `VariableEquals returns false when key missing`() { + val ctx = contextWith() + assertFalse(VariableEquals("status", "done").evaluate(ctx)) + } + + @Test + fun `VariableEquals returns false when value differs`() { + val ctx = contextWith(variables = mapOf("status" to "pending")) + assertFalse(VariableEquals("status", "done").evaluate(ctx)) + } + + // AllOf + + @Test + fun `AllOf with empty list returns true (vacuous truth)`() { + assertTrue(AllOf(emptyList()).evaluate(baseContext)) + } + + @Test + fun `AllOf with all true conditions returns true`() { + assertTrue(AllOf(listOf(AlwaysTrue, AlwaysTrue)).evaluate(baseContext)) + } + + @Test + fun `AllOf with one false condition returns false`() { + val alwaysFalse = Not(AlwaysTrue) + assertFalse(AllOf(listOf(AlwaysTrue, alwaysFalse)).evaluate(baseContext)) + } + + // AnyOf + + @Test + fun `AnyOf with empty list returns false (vacuous falsity)`() { + assertFalse(AnyOf(emptyList()).evaluate(baseContext)) + } + + @Test + fun `AnyOf with one true condition returns true`() { + val alwaysFalse = Not(AlwaysTrue) + assertTrue(AnyOf(listOf(alwaysFalse, AlwaysTrue)).evaluate(baseContext)) + } + + @Test + fun `AnyOf with all false conditions returns false`() { + val alwaysFalse = Not(AlwaysTrue) + assertFalse(AnyOf(listOf(alwaysFalse, alwaysFalse)).evaluate(baseContext)) + } + + // Not + + @Test + fun `Not inverts true to false`() { + assertFalse(Not(AlwaysTrue).evaluate(baseContext)) + } + + @Test + fun `Not inverts false to true`() { + assertTrue(Not(Not(AlwaysTrue)).evaluate(baseContext)) + } +} diff --git a/infrastructure/build.gradle b/infrastructure/build.gradle index ce1c3139..90c5e05b 100644 --- a/infrastructure/build.gradle +++ b/infrastructure/build.gradle @@ -16,6 +16,8 @@ dependencies { implementation project(":infrastructure:persistence") implementation project(":infrastructure:tools") implementation project(":infrastructure:tools:filesystem") + implementation project(":infrastructure:workflow") + implementation project(":core:artifacts") 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" diff --git a/infrastructure/persistence/build.gradle b/infrastructure/persistence/build.gradle index 67a575b7..17907e75 100644 --- a/infrastructure/persistence/build.gradle +++ b/infrastructure/persistence/build.gradle @@ -8,6 +8,7 @@ dependencies { implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version" implementation(project(":core:events")) implementation(project(":core:sessions")) + implementation(project(":core:artifacts")) implementation "org.xerial:sqlite-jdbc" testImplementation(testFixtures(project(":testing:contracts"))) testImplementation "org.junit.jupiter:junit-jupiter" diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt new file mode 100644 index 00000000..7288a529 --- /dev/null +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/artifact/LiveArtifactRepository.kt @@ -0,0 +1,96 @@ +package com.correx.infrastructure.persistence.artifact + +import com.correx.core.artifacts.ArtifactProjector +import com.correx.core.artifacts.ArtifactReducer +import com.correx.core.artifacts.ArtifactState +import com.correx.core.artifacts.repository.ArtifactRepository +import com.correx.core.events.events.ArtifactArchivedEvent +import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.ArtifactRejectedEvent +import com.correx.core.events.events.ArtifactRelationshipAddedEvent +import com.correx.core.events.events.ArtifactSupersededEvent +import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.ArtifactValidatingEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.flow.collect +import kotlinx.coroutines.launch +import java.util.concurrent.ConcurrentHashMap + +class LiveArtifactRepository( + private val eventStore: EventStore, + private val reducer: ArtifactReducer, + private val scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()), +) : ArtifactRepository { + + private val artifactCache = ConcurrentHashMap>() + private val stageIndex = ConcurrentHashMap>() + private val subscriptions = ConcurrentHashMap() + private val projector = ArtifactProjector(reducer) + + override fun getBySession(sessionId: SessionId): Map { + ensureSubscribed(sessionId) + return artifactCache[sessionId]?.toMap() ?: emptyMap() + } + + override fun getByStage(sessionId: SessionId, stageId: StageId): Map { + ensureSubscribed(sessionId) + val cache = artifactCache[sessionId] ?: emptyMap() + val index = stageIndex[sessionId] ?: emptyMap() + return cache.filter { (artifactId, _) -> index[artifactId] == stageId } + } + + override fun getById(sessionId: SessionId, artifactId: ArtifactId): ArtifactState? { + ensureSubscribed(sessionId) + return artifactCache[sessionId]?.get(artifactId) + } + + override fun getByIds(sessionId: SessionId, artifactIds: Set): Map { + ensureSubscribed(sessionId) + val cache = artifactCache[sessionId] ?: return emptyMap() + return artifactIds.mapNotNull { id -> cache[id]?.let { id to it } }.toMap() + } + + private fun ensureSubscribed(sessionId: SessionId) { + subscriptions.computeIfAbsent(sessionId) { + scope.launch { + eventStore.subscribe(sessionId).collect { event -> + processEvent(sessionId, event) + } + } + } + } + + private fun processEvent(sessionId: SessionId, event: StoredEvent) { + val artifactId = extractArtifactId(event) ?: return + val sessionArtifacts = artifactCache.computeIfAbsent(sessionId) { ConcurrentHashMap() } + val currentState = sessionArtifacts[artifactId] ?: projector.initial() + val newState = projector.apply(currentState, event) + sessionArtifacts[artifactId] = newState + + val payload = event.payload + if (payload is ArtifactCreatedEvent) { + val sessionStages = stageIndex.computeIfAbsent(sessionId) { ConcurrentHashMap() } + sessionStages[artifactId] = payload.stageId + } + } + + private fun extractArtifactId(event: StoredEvent): ArtifactId? = + when (val p = event.payload) { + is ArtifactCreatedEvent -> p.artifactId + is ArtifactValidatingEvent -> p.artifactId + is ArtifactValidatedEvent -> p.artifactId + is ArtifactRejectedEvent -> p.artifactId + is ArtifactSupersededEvent -> p.artifactId + is ArtifactArchivedEvent -> p.artifactId + is ArtifactRelationshipAddedEvent -> p.sourceId + else -> null + } +} diff --git a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt index 951289e4..a43f4ef4 100644 --- a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt +++ b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt @@ -12,21 +12,37 @@ import com.correx.core.tools.contract.ToolExecutor import com.correx.core.tools.registry.ToolRegistry import com.correx.infrastructure.inference.DefaultProviderRegistry import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy +import com.correx.infrastructure.inference.commons.ModelDescriptor import com.correx.infrastructure.inference.commons.ModelManager +import com.correx.infrastructure.inference.commons.ResidencyMode import com.correx.infrastructure.inference.llama.cpp.DefaultModelManager +import com.correx.infrastructure.inference.llama.cpp.LlamaCppInferenceProvider import com.correx.infrastructure.persistence.SqliteEventStore import com.correx.infrastructure.tools.DefaultToolRegistry import com.correx.infrastructure.tools.DispatchingToolExecutor import com.correx.infrastructure.tools.SandboxedToolExecutor import com.correx.infrastructure.tools.ToolConfig import com.correx.infrastructure.tools.buildTools +import com.correx.infrastructure.workflow.FileSystemPromptLoader +import com.correx.infrastructure.workflow.PromptLoader +import com.correx.infrastructure.workflow.TomlWorkflowLoader +import com.correx.infrastructure.workflow.WorkflowLoader +import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository +import com.correx.core.artifacts.repository.ArtifactRepository +import com.correx.core.artifacts.DefaultArtifactReducer import io.ktor.client.HttpClient import java.nio.file.Path import java.sql.DriverManager object InfrastructureModule { - fun createEventStore(dbPath: String): EventStore = - SqliteEventStore(DriverManager.getConnection("jdbc:sqlite:$dbPath")) + private val defaultDbPath: String = + "${System.getProperty("user.home")}/.config/correx/correx.db" + + fun createEventStore(dbPath: String = defaultDbPath): EventStore { + val dir = java.io.File(dbPath).parentFile + if (!dir.exists()) dir.mkdirs() + return SqliteEventStore(DriverManager.getConnection("jdbc:sqlite:$dbPath")) + } fun createModelManager( eventStore: EventStore, @@ -47,6 +63,31 @@ object InfrastructureModule { strategy: RoutingStrategy = FirstAvailableRoutingStrategy(), ): InferenceRouter = DefaultInferenceRouter(registry, strategy) + fun createLlamaCppProvider( + modelId: String, + modelPath: String, + baseUrl: String = "http://127.0.0.1:10000", + residencyMode: ResidencyMode = ResidencyMode.PERSISTENT, + ): LlamaCppInferenceProvider = LlamaCppInferenceProvider( + descriptor = ModelDescriptor( + modelId = modelId, + modelPath = modelPath, + residencyMode = residencyMode, + ), + baseUrl = baseUrl, + ) + + fun createProviderRegistry( + providers: List = emptyList(), + ): DefaultProviderRegistry = DefaultProviderRegistry(providers) + + fun createArtifactRepository(eventStore: EventStore): ArtifactRepository = + LiveArtifactRepository(eventStore, DefaultArtifactReducer()) + + fun createWorkflowLoader(): WorkflowLoader = TomlWorkflowLoader() + + fun createPromptLoader(): PromptLoader = FileSystemPromptLoader() + fun createToolExecutor( registry: ToolRegistry, approvalEngine: ApprovalEngine, diff --git a/infrastructure/workflow/build.gradle b/infrastructure/workflow/build.gradle new file mode 100644 index 00000000..8fda310e --- /dev/null +++ b/infrastructure/workflow/build.gradle @@ -0,0 +1,14 @@ +plugins { + id 'java-library' + id 'org.jetbrains.kotlin.jvm' +} + +dependencies { + implementation(project(":core:transitions")) + implementation(project(":core:inference")) + implementation(project(":core:events")) + implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-toml:2.17.0") + implementation("com.fasterxml.jackson.module:jackson-module-kotlin:2.17.0") + testImplementation "org.junit.jupiter:junit-jupiter" + testImplementation "org.jetbrains.kotlin:kotlin-test-junit5" +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionFactory.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionFactory.kt new file mode 100644 index 00000000..01d096a5 --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionFactory.kt @@ -0,0 +1,39 @@ +package com.correx.infrastructure.workflow + +import com.correx.core.events.types.ArtifactId +import com.correx.core.transitions.conditions.AllOf +import com.correx.core.transitions.conditions.AlwaysTrue +import com.correx.core.transitions.conditions.AnyOf +import com.correx.core.transitions.conditions.ArtifactAbsent +import com.correx.core.transitions.conditions.ArtifactPresent +import com.correx.core.transitions.conditions.ArtifactValidated +import com.correx.core.transitions.conditions.Not +import com.correx.core.transitions.conditions.VariableEquals +import com.correx.core.transitions.graph.TransitionCondition + +fun ConditionSpec.toCondition(): TransitionCondition = when (type) { + "always_true" -> AlwaysTrue + "artifact_present" -> buildArtifactPresent() + "artifact_absent" -> buildArtifactAbsent() + "artifact_validated" -> buildArtifactValidated() + "variable_equals" -> buildVariableEquals() + "all_of" -> AllOf(conditions.map { it.toCondition() }) + "any_of" -> AnyOf(conditions.map { it.toCondition() }) + "not" -> Not((condition ?: error("condition required for not")).toCondition()) + else -> throw WorkflowValidationException("unknown condition type: $type") +} + +private fun ConditionSpec.buildArtifactPresent() = + ArtifactPresent(ArtifactId(artifactId ?: error("artifact_id required for artifact_present"))) + +private fun ConditionSpec.buildArtifactAbsent() = + ArtifactAbsent(ArtifactId(artifactId ?: error("artifact_id required for artifact_absent"))) + +private fun ConditionSpec.buildArtifactValidated() = + ArtifactValidated(ArtifactId(artifactId ?: error("artifact_id required for artifact_validated"))) + +private fun ConditionSpec.buildVariableEquals() = + VariableEquals( + key ?: error("key required for variable_equals"), + value ?: error("value required for variable_equals"), + ) diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionSpec.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionSpec.kt new file mode 100644 index 00000000..6245436e --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/ConditionSpec.kt @@ -0,0 +1,10 @@ +package com.correx.infrastructure.workflow + +data class ConditionSpec( + val type: String, + val artifactId: String? = null, + val key: String? = null, + val value: String? = null, + val conditions: List = emptyList(), + val condition: ConditionSpec? = null, +) diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoader.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoader.kt new file mode 100644 index 00000000..2df760ed --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoader.kt @@ -0,0 +1,19 @@ +package com.correx.infrastructure.workflow + +import java.nio.file.Path +import kotlin.io.path.exists +import kotlin.io.path.readText + +class FileSystemPromptLoader( + private val configDir: Path = Path.of(System.getProperty("user.home"), ".config", "correx", "prompts"), +) : PromptLoader { + override fun load(path: String): String { + val local = Path.of(path) + if (local.exists()) return local.readText() + + val fromConfig = configDir.resolve(path) + if (fromConfig.exists()) return fromConfig.readText() + + throw PromptNotFoundException("Prompt not found: $path (searched: $local, $fromConfig)") + } +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptLoader.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptLoader.kt new file mode 100644 index 00000000..c08c6470 --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptLoader.kt @@ -0,0 +1,5 @@ +package com.correx.infrastructure.workflow + +interface PromptLoader { + fun load(path: String): String +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptNotFoundException.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptNotFoundException.kt new file mode 100644 index 00000000..8f987d43 --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/PromptNotFoundException.kt @@ -0,0 +1,3 @@ +package com.correx.infrastructure.workflow + +class PromptNotFoundException(message: String) : Exception(message) diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt new file mode 100644 index 00000000..41b8ef86 --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt @@ -0,0 +1,130 @@ +package com.correx.infrastructure.workflow + +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.TransitionId +import com.correx.core.transitions.graph.StageConfig +import com.correx.core.transitions.graph.TransitionEdge +import com.correx.core.transitions.graph.WorkflowGraph +import com.fasterxml.jackson.annotation.JsonProperty +import com.fasterxml.jackson.databind.DeserializationFeature +import com.fasterxml.jackson.dataformat.toml.TomlMapper +import com.fasterxml.jackson.module.kotlin.kotlinModule +import com.fasterxml.jackson.module.kotlin.readValue +import java.nio.file.Path +import kotlin.io.path.readText + +private data class WorkflowFile( + val id: String = "", + val start: String = "", + val stages: List = emptyList(), + val transitions: List = emptyList(), +) + +private data class StageSection( + val id: String = "", + val prompt: String? = null, + val produces: List = emptyList(), + val needs: List = emptyList(), + @JsonProperty("allowed_tools") val allowedTools: List = emptyList(), + @JsonProperty("token_budget") val tokenBudget: Int = 4096, + @JsonProperty("max_retries") val maxRetries: Int = 3, +) + +// condition fields flattened into the transition row +private data class TransitionSection( + val id: String = "", + val from: String = "", + val to: String = "", + @JsonProperty("condition_type") val conditionType: String = "", + @JsonProperty("condition_artifact_id") val conditionArtifactId: String? = null, + @JsonProperty("condition_key") val conditionKey: String? = null, + @JsonProperty("condition_value") val conditionValue: String? = null, +) + +private const val TERMINAL = "done" + +private val mapper = TomlMapper.builder() + .addModule(kotlinModule()) + .disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES) + .build() + +class TomlWorkflowLoader : WorkflowLoader { + override fun load(path: Path): WorkflowGraph { + val raw = path.readText() + val file = runCatching { mapper.readValue(raw) } + .getOrElse { throw WorkflowValidationException("Failed to parse workflow TOML: ${it.message}") } + return file.toWorkflowGraph() + } + + private fun WorkflowFile.toWorkflowGraph(): WorkflowGraph { + val startId = StageId(start) + val stageMap = stages.associate { s -> + StageId(s.id) to StageConfig( + produces = s.produces.map { ArtifactId(it) }.toSet(), + needs = s.needs.map { ArtifactId(it) }.toSet(), + allowedTools = s.allowedTools.toSet(), + tokenBudget = s.tokenBudget, + maxRetries = s.maxRetries, + metadata = buildMap { s.prompt?.let { put("prompt", it) } }, + ) + } + + val declaredIds = stageMap.keys.map { it.value }.toSet() + validate(start, declaredIds, transitions) + + val allProduced = stageMap.values.flatMap { it.produces }.toSet() + stageMap.forEach { (stageId, config) -> + config.needs.forEach { needed -> + if (needed !in allProduced) { + throw WorkflowValidationException( + "Stage '${stageId.value}' needs artifact '${needed.value}' but no stage produces it" + ) + } + } + } + + val transitionSet = transitions.map { t -> + TransitionEdge( + id = TransitionId(t.id), + from = StageId(t.from), + to = StageId(t.to), + condition = ConditionSpec( + type = t.conditionType, + artifactId = t.conditionArtifactId, + key = t.conditionKey, + value = t.conditionValue, + ).toCondition(), + ) + }.toSet() + + return WorkflowGraph(id = id, stages = stageMap, transitions = transitionSet, start = startId) + } + + @Suppress("ThrowsCount") + private fun validate( + start: String, + declaredStages: Set, + transitions: List, + ) { + if (start !in declaredStages) { + throw WorkflowValidationException("start stage '$start' is not declared in stages") + } + val seenIds = mutableSetOf() + transitions.forEach { t -> + if (!seenIds.add(t.id)) { + throw WorkflowValidationException("duplicate transition id: '${t.id}'") + } + checkTransitionStage(t, "from", t.from, declaredStages) + checkTransitionStage(t, "to", t.to, declaredStages) + } + } + + private fun checkTransitionStage(t: TransitionSection, field: String, stageRef: String, declared: Set) { + if (stageRef !in declared && (field != "to" || stageRef != TERMINAL)) { + throw WorkflowValidationException( + "transition '${t.id}' references unknown $field-stage '$stageRef'" + ) + } + } +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowLoader.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowLoader.kt new file mode 100644 index 00000000..1c4aa2df --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowLoader.kt @@ -0,0 +1,8 @@ +package com.correx.infrastructure.workflow + +import com.correx.core.transitions.graph.WorkflowGraph +import java.nio.file.Path + +interface WorkflowLoader { + fun load(path: Path): WorkflowGraph +} diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowValidationException.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowValidationException.kt new file mode 100644 index 00000000..738c47f6 --- /dev/null +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/WorkflowValidationException.kt @@ -0,0 +1,3 @@ +package com.correx.infrastructure.workflow + +class WorkflowValidationException(message: String) : Exception(message) diff --git a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoaderTest.kt b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoaderTest.kt new file mode 100644 index 00000000..35dd1f08 --- /dev/null +++ b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/FileSystemPromptLoaderTest.kt @@ -0,0 +1,45 @@ +package com.correx.infrastructure.workflow + +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import java.nio.file.Files +import kotlin.io.path.writeText +import kotlin.test.assertEquals +import kotlin.test.assertContains + +class FileSystemPromptLoaderTest { + + @Test + fun `loads from absolute path when file exists`() { + val file = Files.createTempFile("prompt", ".md").also { it.writeText("hello world") } + val loader = FileSystemPromptLoader() + assertEquals("hello world", loader.load(file.toAbsolutePath().toString())) + } + + @Test + fun `falls back to configDir when local path not found`() { + val configDir = Files.createTempDirectory("correx-prompts") + configDir.resolve("test.md").also { it.writeText("from config") } + val loader = FileSystemPromptLoader(configDir = configDir) + assertEquals("from config", loader.load("test.md")) + } + + @Test + fun `throws PromptNotFoundException with both paths when not found`() { + val configDir = Files.createTempDirectory("correx-prompts-empty") + val loader = FileSystemPromptLoader(configDir = configDir) + val ex = assertThrows { loader.load("missing.md") } + assertContains(ex.message!!, "missing.md") + assertContains(ex.message!!, configDir.toString()) + } + + @Test + fun `local path takes precedence over configDir`() { + val localFile = Files.createTempFile("prompt", ".md").also { it.writeText("local") } + val configDir = Files.createTempDirectory("correx-prompts") + configDir.resolve(localFile.fileName.toString()).also { it.writeText("from config") } + + val loader = FileSystemPromptLoader(configDir = configDir) + assertEquals("local", loader.load(localFile.toAbsolutePath().toString())) + } +} diff --git a/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoaderTest.kt b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoaderTest.kt new file mode 100644 index 00000000..5c26bab5 --- /dev/null +++ b/infrastructure/workflow/src/test/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoaderTest.kt @@ -0,0 +1,93 @@ +package com.correx.infrastructure.workflow + +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows +import java.nio.file.Files +import kotlin.io.path.writeText +import kotlin.test.assertEquals + +class TomlWorkflowLoaderTest { + + private val loader = TomlWorkflowLoader() + + private val validToml = """ + id = "healthcheck" + start = "collect" + + [[stages]] + id = "collect" + prompt = "prompts/collect.md" + produces = ["system_stats"] + allowed_tools = ["ShellTool"] + token_budget = 2048 + max_retries = 2 + + [[stages]] + id = "report" + prompt = "prompts/report.md" + needs = ["system_stats"] + produces = ["report"] + token_budget = 2048 + + [[transitions]] + id = "collect-to-report" + from = "collect" + to = "report" + condition_type = "artifact_present" + condition_artifact_id = "system_stats" + + [[transitions]] + id = "report-to-done" + from = "report" + to = "done" + condition_type = "always_true" + """.trimIndent() + + @Test + fun `valid toml produces correct WorkflowGraph`() { + val path = Files.createTempFile("workflow", ".toml").also { it.writeText(validToml) } + val graph = loader.load(path) + + assertEquals("collect", graph.start.value) + assertEquals(2, graph.stages.size) + assertEquals(2, graph.transitions.size) + + val collectStage = graph.stages[graph.start]!! + assertEquals(setOf("system_stats"), collectStage.produces.map { it.value }.toSet()) + assertEquals("prompts/collect.md", collectStage.metadata["prompt"]) + + val reportStage = graph.stages.values.first { it.produces.any { a -> a.value == "report" } } + assertEquals(setOf("system_stats"), reportStage.needs.map { it.value }.toSet()) + } + + @Test + fun `missing stage reference throws WorkflowValidationException`() { + val bad = validToml.replace("from = \"collect\"", "from = \"nonexistent\"") + val path = Files.createTempFile("workflow", ".toml").also { it.writeText(bad) } + assertThrows { loader.load(path) } + } + + @Test + fun `duplicate transition ids throw WorkflowValidationException`() { + val bad = validToml.replace("id = \"report-to-done\"", "id = \"collect-to-report\"") + val path = Files.createTempFile("workflow", ".toml").also { it.writeText(bad) } + assertThrows { loader.load(path) } + } + + @Test + fun `unknown start stage throws WorkflowValidationException`() { + val bad = validToml.replace("start = \"collect\"", "start = \"missing\"") + val path = Files.createTempFile("workflow", ".toml").also { it.writeText(bad) } + assertThrows { loader.load(path) } + } + + @Test + fun `unsatisfied needs throws WorkflowValidationException`() { + val bad = validToml.replace( + "needs = [\"system_stats\"]", + "needs = [\"nonexistent_artifact\"]", + ) + val path = Files.createTempFile("workflow", ".toml").also { it.writeText(bad) } + assertThrows { loader.load(path) } + } +} diff --git a/prompts/healthcheck_execute.md b/prompts/healthcheck_execute.md new file mode 100644 index 00000000..898ebc3b --- /dev/null +++ b/prompts/healthcheck_execute.md @@ -0,0 +1,4 @@ +# healthcheck_execute.md + +Run `scripts/healthcheck.sh` and report the results. +If the script fails, report the error and exit code. \ No newline at end of file diff --git a/prompts/healthcheck_write.md b/prompts/healthcheck_write.md new file mode 100644 index 00000000..1d0661d9 --- /dev/null +++ b/prompts/healthcheck_write.md @@ -0,0 +1,10 @@ +# healthcheck_write.md + +Write a bash script to `scripts/healthcheck.sh` that reports: +- RAM usage % +- CPU usage % +- GPU usage % and VRAM usage % if an NVIDIA or AMD GPU is detected + +Use standard tools only (free, top/vmstat, nvidia-smi, rocm-smi). +If no GPU is detected, skip that section gracefully. +Make the script executable. \ No newline at end of file diff --git a/settings.gradle b/settings.gradle index f8ce649f..13e418c8 100644 --- a/settings.gradle +++ b/settings.gradle @@ -40,6 +40,7 @@ include ':infrastructure:tools:filesystem' include ':infrastructure:security' include ':infrastructure:scheduler' include ':infrastructure:telemetry' +include ':infrastructure:workflow' include ':interfaces:api' include ':interfaces:cli' diff --git a/testing/integration/build.gradle b/testing/integration/build.gradle index 37193e65..7416239a 100644 --- a/testing/integration/build.gradle +++ b/testing/integration/build.gradle @@ -15,6 +15,7 @@ dependencies { testImplementation(project(":core:kernel")) testImplementation(project(":core:risk")) testImplementation(project(":infrastructure:persistence")) + testImplementation(project(":core:artifacts")) testImplementation(project(":testing:fixtures")) testImplementation(project(":testing:kernel")) testImplementation(project(":testing:contracts")) diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index 79aa8aac..186b5a9f 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -35,7 +35,9 @@ import com.correx.core.transitions.graph.StageConfig import com.correx.core.transitions.graph.WorkflowGraph import com.correx.core.validation.approval.ApprovalTrigger import com.correx.core.validation.pipeline.ValidationPipeline +import com.correx.core.artifacts.DefaultArtifactReducer import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository import com.correx.testing.fixtures.InferenceFixtures import com.correx.testing.fixtures.context.ContextFixtures import com.correx.testing.fixtures.cyclePolicyMissingValidator @@ -82,6 +84,7 @@ class SessionOrchestratorIntegrationTest { inferenceRepository = inferenceRepository, orchestrationRepository = orchestrationRepository, sessionRepository = sessionRepository, + artifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()), ) private val engines = OrchestratorEngines(