chore(damn): detekt, build, tests, formatting

fixed detekt issues where possible.
fixed disttar failing build because tools is added twice in the server module.
added workflowId where required.
fixed some tests not being recognized because of runBlocking without explicit return type.
formatting + imports.
This commit is contained in:
2026-05-22 00:10:05 +04:00
parent 2c459da009
commit f827685ed0
185 changed files with 1134 additions and 832 deletions
+7
View File
@@ -8,6 +8,13 @@ application {
mainClass = 'com.correx.apps.server.MainKt'
}
tasks.named('distTar') {
duplicatesStrategy = DuplicatesStrategy.EXCLUDE
}
tasks.named('distZip') {
duplicatesStrategy = DuplicatesStrategy.EXCLUDE
}
dependencies {
implementation project(':core:config')
implementation project(':core:events')
@@ -9,7 +9,6 @@ import io.ktor.serialization.kotlinx.json.json
import io.ktor.server.application.Application
import io.ktor.server.application.install
import io.ktor.server.plugins.calllogging.CallLogging
import org.slf4j.event.Level
import io.ktor.server.plugins.contentnegotiation.ContentNegotiation
import io.ktor.server.plugins.statuspages.StatusPages
import io.ktor.server.response.respond
@@ -18,6 +17,7 @@ import io.ktor.server.routing.routing
import io.ktor.server.websocket.WebSockets
import io.ktor.server.websocket.webSocket
import kotlinx.serialization.json.Json
import org.slf4j.event.Level
fun Application.configureServer(module: ServerModule) {
install(CallLogging) { level = Level.INFO }
@@ -49,9 +49,20 @@ import java.nio.file.Paths
private val log = LoggerFactory.getLogger("com.correx.apps.server.Main")
fun main() {
log.info("=== correx server starting ===")
log.info(" port : 8080")
val artifactStore = InfrastructureModule.createArtifactStore()
val eventStore = LoggingEventStore(InfrastructureModule.createEventStore(artifactStore))
logStoresInfo(eventStore, artifactStore)
val infraRegistry = InfrastructureModule.createProviderRegistry(listOf(buildLlamaProvider()))
val modelId = System.getenv("CORREX_MODEL_ID") ?: "default"
val modelPath = System.getenv("CORREX_MODEL_PATH") ?: "(not set)"
val llamaUrl = System.getenv("CORREX_LLAMA_URL") ?: "http://127.0.0.1:10000"
logModelInfo(modelId, modelPath, llamaUrl, infraRegistry)
val repositories = buildRepositories(eventStore)
val approvalEngine = DefaultApprovalEngine()
val correxConfig = ConfigLoader.load()
@@ -78,6 +89,9 @@ fun main() {
eventDispatcher = EventDispatcher(eventStore),
workDir = sandboxRoot,
)
logToolInfo(sandboxRoot, workingDir, shellAllowedExecutables, toolRegistry)
val inferenceRouter = DefaultInferenceRouter(infraRegistry, FirstAvailableRoutingStrategy())
val engines = OrchestratorEngines(
transitionResolver = DefaultTransitionResolver { condition, ctx -> condition.evaluate(ctx) },
@@ -116,52 +130,46 @@ fun main() {
defaultOrchestrationConfig = defaultOrchestrationConfig,
routerFacade = routerFacade,
)
val modelId = System.getenv("CORREX_MODEL_ID") ?: "default"
val modelPath = System.getenv("CORREX_MODEL_PATH") ?: "(not set)"
val llamaUrl = System.getenv("CORREX_LLAMA_URL") ?: "http://127.0.0.1:10000"
logServerStartup(
modelId,
modelPath,
llamaUrl,
sandboxRoot,
workingDir,
shellAllowedExecutables,
eventStore,
artifactStore,
toolRegistry,
infraRegistry,
)
log.info("==============================")
embeddedServer(Netty, port = 8080) { configureServer(module) }.start(wait = true)
}
private fun logServerStartup(
private fun logModelInfo(
modelId: String,
modelPath: String,
llamaUrl: String,
sandboxRoot: Path?,
workingDir: Path?,
shellAllowedExecutables: Set<String>,
eventStore: LoggingEventStore,
artifactStore: ArtifactStore,
toolRegistry: ToolRegistry,
infraRegistry: DefaultProviderRegistry,
) {
log.info("=== correx server starting ===")
log.info(" port : 8080")
log.info("==========MODEL INFO==========")
log.info(" model id : {}", modelId)
log.info(" model path : {}", modelPath)
log.info(" llama url : {}", llamaUrl)
log.info(" providers : {} registered", infraRegistry.listAll().size)
}
private fun logToolInfo(
sandboxRoot: Path?,
workingDir: Path?,
shellAllowedExecutables: Set<String>,
toolRegistry: ToolRegistry,
) {
log.info("==========TOOLS INFO==========")
log.info(" sandbox root : {}", sandboxRoot)
log.info(" working dir : {}", workingDir)
if (shellAllowedExecutables.isEmpty()) {
log.warn(" shell tool : no executable allowlist configured — all executables allowed")
}
log.info(" tools : {}", toolRegistry.all().map { it.name })
}
private fun logStoresInfo(
eventStore: LoggingEventStore,
artifactStore: ArtifactStore,
) {
log.info("==========STORE INFO==========")
log.info(" event store : {}", eventStore::class.simpleName)
log.info(" artifact store: {}", artifactStore::class.simpleName)
log.info(" tools : {}", toolRegistry.all().map { it.name })
log.info(" providers : {} registered", infraRegistry.listAll().size)
log.info("==============================")
}
private fun buildLlamaProvider(): LlamaCppInferenceProvider = InfrastructureModule.createLlamaCppProvider(
@@ -6,8 +6,8 @@ import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.stores.EventStore
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
import com.correx.core.kernel.orchestration.OrchestrationConfig
import com.correx.core.sessions.DefaultSessionRepository
import com.correx.core.router.RouterFacade
import com.correx.core.sessions.DefaultSessionRepository
data class ServerModule(
val orchestrator: DefaultSessionOrchestrator,
@@ -10,7 +10,6 @@ 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
@@ -25,7 +24,8 @@ import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.datetime.Clock
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.*
import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision
class ApprovalCoordinator(
private val orchestrator: ApprovalGateway,
@@ -5,13 +5,13 @@ import com.correx.apps.server.protocol.RiskSummaryDto
import com.correx.apps.server.protocol.ServerMessage
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.events.InferenceCompletedEvent
import com.correx.core.events.events.InferenceStartedEvent
import com.correx.core.events.events.InferenceTimeoutEvent
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.events.ToolExecutionCompletedEvent
import com.correx.core.events.events.ToolExecutionFailedEvent
import com.correx.core.events.events.ToolExecutionRejectedEvent
@@ -20,7 +20,7 @@ import com.correx.core.events.events.TransitionExecutedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.types.ArtifactId
class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) {
suspend fun map(event: StoredEvent): ServerMessage? =
@@ -39,47 +39,80 @@ suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: Artifa
is WorkflowStartedEvent -> mapWorkflowStarted(p)
is WorkflowCompletedEvent -> ServerMessage.SessionCompleted(sessionId = p.sessionId)
is WorkflowFailedEvent -> ServerMessage.SessionFailed(sessionId = p.sessionId, reason = p.reason)
is TransitionExecutedEvent -> ServerMessage.StageStarted(sessionId = p.sessionId, stageId = p.to)
is StageCompletedEvent -> ServerMessage.StageCompleted(sessionId = p.sessionId, stageId = p.stageId)
is StageFailedEvent -> ServerMessage.StageFailed(
sessionId = p.sessionId, stageId = p.stageId, reason = p.reason,
is TransitionExecutedEvent -> ServerMessage.StageStarted(
sessionId = p.sessionId,
stageId = p.to,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
is StageCompletedEvent -> ServerMessage.StageCompleted(
sessionId = p.sessionId,
stageId = p.stageId,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
is StageFailedEvent -> ServerMessage.StageFailed(
sessionId = p.sessionId,
stageId = p.stageId,
reason = p.reason,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
is OrchestrationPausedEvent -> mapOrchestrationPaused(p)
is InferenceStartedEvent -> ServerMessage.InferenceStarted(sessionId = p.sessionId, stageId = p.stageId)
is InferenceCompletedEvent -> mapInferenceCompleted(p, artifactStore)
is InferenceCompletedEvent -> mapInferenceCompleted(event, p, artifactStore)
is InferenceTimeoutEvent -> ServerMessage.InferenceTimedOut(
sessionId = p.sessionId, stageId = p.stageId, elapsedMs = p.timeoutMs,
)
is ToolInvocationRequestedEvent -> ServerMessage.ToolStarted(
sessionId = p.sessionId, toolName = p.toolName, tier = p.tier,
)
is ToolExecutionCompletedEvent -> ServerMessage.ToolCompleted(
sessionId = p.sessionId, toolName = p.toolName, outputSummary = p.receipt.outputSummary,
sessionId = p.sessionId,
toolName = p.toolName,
outputSummary = p.receipt.outputSummary,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
is ToolExecutionFailedEvent -> ServerMessage.ToolFailed(
sessionId = p.sessionId, toolName = p.toolName, reason = p.reason,
sessionId = p.sessionId,
toolName = p.toolName,
reason = p.reason,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
is ToolExecutionRejectedEvent -> ServerMessage.ToolRejected(
sessionId = p.sessionId, toolName = p.toolName, reason = p.reason,
)
is ApprovalRequestedEvent -> mapApprovalRequested(p)
else -> null
}
private fun mapWorkflowStarted(p: WorkflowStartedEvent): ServerMessage =
ServerMessage.SessionStarted(sessionId = p.sessionId, workflowId = p.startStageId.value)
ServerMessage.SessionStarted(sessionId = p.sessionId, workflowId = p.workflowId)
private fun mapOrchestrationPaused(p: OrchestrationPausedEvent): ServerMessage {
val reason = if (p.reason == "APPROVAL_PENDING") PauseReason.APPROVAL_PENDING else PauseReason.USER_REQUESTED
return ServerMessage.SessionPaused(sessionId = p.sessionId, reason = reason)
}
private suspend fun mapInferenceCompleted(p: InferenceCompletedEvent, artifactStore: ArtifactStore): ServerMessage {
private suspend fun mapInferenceCompleted(
event: StoredEvent,
p: InferenceCompletedEvent,
artifactStore: ArtifactStore,
): ServerMessage {
val text = runCatching {
artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: ""
}.getOrElse { "" }
return ServerMessage.InferenceCompleted(
sessionId = p.sessionId, stageId = p.stageId, outputSummary = text, responseText = text,
sessionId = p.sessionId,
stageId = p.stageId,
outputSummary = text,
responseText = text,
occurredAt = event.metadata.timestamp.toEpochMilliseconds(),
)
}
@@ -1,7 +1,7 @@
package com.correx.apps.server.protocol
import kotlinx.serialization.json.Json
import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json
class ProtocolException(message: String, cause: Throwable? = null) : RuntimeException(message, cause)
@@ -21,13 +21,13 @@ sealed class ServerMessage {
data class SessionFailed(val sessionId: SessionId, val reason: String) : ServerMessage()
@Serializable
data class StageStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage()
data class StageStarted(val sessionId: SessionId, val stageId: StageId, val occurredAt: Long) : ServerMessage()
@Serializable
data class StageCompleted(val sessionId: SessionId, val stageId: StageId) : ServerMessage()
data class StageCompleted(val sessionId: SessionId, val stageId: StageId, val occurredAt: Long) : ServerMessage()
@Serializable
data class StageFailed(val sessionId: SessionId, val stageId: StageId, val reason: String) :
data class StageFailed(val sessionId: SessionId, val stageId: StageId, val reason: String, val occurredAt: Long) :
ServerMessage()
@Serializable
@@ -39,6 +39,7 @@ sealed class ServerMessage {
val stageId: StageId,
val outputSummary: String,
val responseText: String = "",
val occurredAt: Long,
) : ServerMessage()
@Serializable
@@ -50,11 +51,15 @@ sealed class ServerMessage {
ServerMessage()
@Serializable
data class ToolCompleted(val sessionId: SessionId, val toolName: String, val outputSummary: String) :
ServerMessage()
data class ToolCompleted(
val sessionId: SessionId,
val toolName: String,
val outputSummary: String,
val occurredAt: Long,
) : ServerMessage()
@Serializable
data class ToolFailed(val sessionId: SessionId, val toolName: String, val reason: String) :
data class ToolFailed(val sessionId: SessionId, val toolName: String, val reason: String, val occurredAt: Long) :
ServerMessage()
@Serializable
@@ -1,8 +1,8 @@
package com.correx.apps.server.registry
import com.correx.core.events.types.ProviderId
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>
@@ -10,7 +10,6 @@ import com.correx.core.events.types.EventId
import com.correx.core.events.types.SessionId
import com.correx.core.utils.TypeId
import io.ktor.http.HttpStatusCode
import org.slf4j.LoggerFactory
import io.ktor.server.request.receive
import io.ktor.server.response.respond
import io.ktor.server.routing.Route
@@ -21,7 +20,8 @@ import io.ktor.server.websocket.webSocket
import kotlinx.coroutines.launch
import kotlinx.datetime.Clock
import kotlinx.serialization.Serializable
import java.util.UUID
import org.slf4j.LoggerFactory
import java.util.*
@Serializable
data class StartSessionRequest(val workflowId: String, val config: SessionConfigDto? = null)
@@ -21,7 +21,7 @@ import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.coroutines.launch
import kotlinx.datetime.Clock
import org.slf4j.LoggerFactory
import java.util.UUID
import java.util.*
private val log = LoggerFactory.getLogger(GlobalStreamHandler::class.java)
@@ -10,12 +10,10 @@ import com.correx.apps.server.protocol.ServerMessage.RouterResponseMessage
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.approvals.model.ApprovalContext
import com.correx.core.approvals.model.ApprovalScopeIdentity
import com.correx.core.events.types.SessionId
import com.correx.core.router.RouterFacade
import com.correx.core.sessions.ApprovalMode
import com.correx.core.utils.TypeId
import io.ktor.server.websocket.DefaultWebSocketServerSession
@@ -24,6 +22,7 @@ import io.ktor.websocket.readText
import kotlinx.coroutines.channels.ClosedReceiveChannelException
import kotlinx.datetime.Clock
import org.slf4j.LoggerFactory
import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision
private val log = LoggerFactory.getLogger(SessionStreamHandler::class.java)
@@ -75,11 +74,20 @@ class SessionStreamHandler(private val module: ServerModule) {
runCatching { module.orchestrator.cancel(msg.sessionId) }
.onFailure { log.error("cancel failed for session={}: {}", msg.sessionId.value, it.message, it) }
}
is ClientMessage.ApprovalResponse -> {
val domainDecision = msg.toDomainDecision(msg.requestId.value)
runCatching { module.orchestrator.submitApprovalDecision(msg.requestId, domainDecision) }
.onFailure { log.error("submitApprovalDecision failed for request={}: {}", msg.requestId.value, it.message, it) }
.onFailure {
log.error(
"submitApprovalDecision failed for request={}: {}",
msg.requestId.value,
it.message,
it,
)
}
}
is ClientMessage.ChatInput -> {
runCatching {
module.routerFacade.onUserInput(
@@ -88,21 +96,24 @@ class SessionStreamHandler(private val module: ServerModule) {
mode = msg.mode,
)
}.onSuccess { response ->
session.send(Frame.Text(
session.send(
Frame.Text(
ProtocolSerializer.encodeServerMessage(
RouterResponseMessage(
sessionId = msg.sessionId,
content = response.content,
steeringEmitted = response.steeringEmitted,
),
),
),
)
)
))
}.onFailure {
log.error("routerFacade.onUserInput failed for session={}: {}", msg.sessionId.value, it.message, it)
val error = ServerMessage.ProtocolError("Router error: ${it.message}")
session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error)))
}
}
else -> {
log.warn("unexpected message type={} in session stream", msg::class.simpleName)
val error = ServerMessage.ProtocolError("Unexpected message type in session stream")
@@ -24,8 +24,8 @@ import com.correx.core.events.events.TransitionExecutedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.ApprovalRequestId
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.EventId
import com.correx.core.events.types.InferenceRequestId
import com.correx.core.events.types.ProviderId
@@ -35,8 +35,8 @@ import com.correx.core.events.types.ToolInvocationId
import com.correx.core.events.types.TransitionId
import com.correx.core.events.types.ValidationReportId
import com.correx.core.inference.TokenUsage
import kotlinx.datetime.Instant
import kotlinx.coroutines.test.runTest
import kotlinx.datetime.Instant
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Test
@@ -45,7 +45,9 @@ class DomainEventMapperTest {
private val sessionId = SessionId("session-1")
private val stageId = StageId("stage-1")
private val workflowId = "workflow-test"
private val timestamp = Instant.parse("2026-01-01T00:00:00Z")
private val occurredAt = timestamp.toEpochMilliseconds()
private val noopStore: ArtifactStore = object : ArtifactStore {
override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("art-1")
override suspend fun get(id: ArtifactId): ByteArray? = null
@@ -68,15 +70,21 @@ class DomainEventMapperTest {
@Test
fun `WorkflowStartedEvent maps to SessionStarted`(): Unit = runTest {
val event = storedEvent(WorkflowStartedEvent(sessionId = sessionId, startStageId = stageId))
val event =
storedEvent(WorkflowStartedEvent(sessionId = sessionId, startStageId = stageId, workflowId = workflowId))
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.SessionStarted(sessionId, stageId.value), result)
assertEquals(ServerMessage.SessionStarted(sessionId, workflowId), result)
}
@Test
fun `WorkflowCompletedEvent maps to SessionCompleted`(): Unit = runTest {
val event = storedEvent(
WorkflowCompletedEvent(sessionId = sessionId, terminalStageId = stageId, totalStages = 1),
WorkflowCompletedEvent(
sessionId = sessionId,
terminalStageId = stageId,
totalStages = 1,
workflowId = workflowId,
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.SessionCompleted(sessionId), result)
@@ -99,7 +107,7 @@ class DomainEventMapperTest {
TransitionExecutedEvent(sessionId = sessionId, from = from, to = to, transitionId = TransitionId("t1")),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.StageStarted(sessionId, to), result)
assertEquals(ServerMessage.StageStarted(sessionId, to, occurredAt), result)
}
@Test
@@ -108,7 +116,7 @@ class DomainEventMapperTest {
StageCompletedEvent(sessionId = sessionId, stageId = stageId, transitionId = TransitionId("t1")),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.StageCompleted(sessionId, stageId), result)
assertEquals(ServerMessage.StageCompleted(sessionId, stageId, occurredAt), result)
}
@Test
@@ -119,7 +127,7 @@ class DomainEventMapperTest {
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.StageFailed(sessionId, stageId, "err"), result)
assertEquals(ServerMessage.StageFailed(sessionId, stageId, "err", occurredAt), result)
}
@Test
@@ -142,13 +150,15 @@ class DomainEventMapperTest {
@Test
fun `InferenceStartedEvent maps to InferenceStarted`(): Unit = runTest {
val event = storedEvent(InferenceStartedEvent(
val event = storedEvent(
InferenceStartedEvent(
requestId = InferenceRequestId("req-1"),
sessionId = sessionId,
stageId = stageId,
providerId = ProviderId("p1"),
promptArtifactId = ArtifactId("art-prompt"),
))
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.InferenceStarted(sessionId, stageId), result)
}
@@ -161,9 +171,11 @@ class DomainEventMapperTest {
override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("x")
override suspend fun get(id: ArtifactId): ByteArray? =
if (id == artifactId) responseText.toByteArray(Charsets.UTF_8) else null
override suspend fun flushBefore(commit: suspend () -> Unit) = commit()
}
val event = storedEvent(InferenceCompletedEvent(
val event = storedEvent(
InferenceCompletedEvent(
requestId = InferenceRequestId("req-1"),
sessionId = sessionId,
stageId = stageId,
@@ -171,14 +183,19 @@ class DomainEventMapperTest {
tokensUsed = TokenUsage(10, 5),
latencyMs = 100L,
responseArtifactId = artifactId,
))
),
)
val result = domainEventToServerMessage(event, store)
assertEquals(ServerMessage.InferenceCompleted(sessionId, stageId, responseText, responseText), result)
assertEquals(
ServerMessage.InferenceCompleted(sessionId, stageId, responseText, responseText, occurredAt),
result,
)
}
@Test
fun `InferenceCompletedEvent falls back to empty string when artifact missing`(): Unit = runTest {
val event = storedEvent(InferenceCompletedEvent(
val event = storedEvent(
InferenceCompletedEvent(
requestId = InferenceRequestId("req-1"),
sessionId = sessionId,
stageId = stageId,
@@ -186,20 +203,23 @@ class DomainEventMapperTest {
tokensUsed = TokenUsage(10, 5),
latencyMs = 100L,
responseArtifactId = ArtifactId("missing"),
))
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.InferenceCompleted(sessionId, stageId, "", ""), result)
assertEquals(ServerMessage.InferenceCompleted(sessionId, stageId, "", "", occurredAt), result)
}
@Test
fun `InferenceTimeoutEvent maps to InferenceTimedOut`(): Unit = runTest {
val event = storedEvent(InferenceTimeoutEvent(
val event = storedEvent(
InferenceTimeoutEvent(
requestId = InferenceRequestId("req-1"),
sessionId = sessionId,
stageId = stageId,
providerId = ProviderId("p1"),
timeoutMs = 5000L,
))
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.InferenceTimedOut(sessionId, stageId, 5000L), result)
}
@@ -207,7 +227,8 @@ class DomainEventMapperTest {
@Test
fun `ToolInvocationRequestedEvent maps to ToolStarted`(): Unit = runTest {
val invId = ToolInvocationId("inv-1")
val event = storedEvent(ToolInvocationRequestedEvent(
val event = storedEvent(
ToolInvocationRequestedEvent(
invocationId = invId,
sessionId = sessionId,
stageId = stageId,
@@ -216,7 +237,8 @@ class DomainEventMapperTest {
request = ToolRequest(
invocationId = invId, sessionId = sessionId, stageId = stageId, toolName = "file_write",
),
))
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.ToolStarted(sessionId, "file_write", Tier.T2), result)
}
@@ -239,7 +261,7 @@ class DomainEventMapperTest {
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.ToolCompleted(sessionId, "file_write", "wrote 3 lines"), result)
assertEquals(ServerMessage.ToolCompleted(sessionId, "file_write", "wrote 3 lines", occurredAt), result)
}
@Test
@@ -253,7 +275,7 @@ class DomainEventMapperTest {
),
)
val result = domainEventToServerMessage(event, noopStore)
assertEquals(ServerMessage.ToolFailed(sessionId, "file_write", "disk full"), result)
assertEquals(ServerMessage.ToolFailed(sessionId, "file_write", "disk full", occurredAt), result)
}
@Test
@@ -274,7 +296,8 @@ class DomainEventMapperTest {
@Test
fun `ApprovalRequestedEvent maps to ApprovalRequired`(): Unit = runTest {
val requestId = ApprovalRequestId("req-1")
val event = storedEvent(ApprovalRequestedEvent(
val event = storedEvent(
ApprovalRequestedEvent(
requestId = requestId,
tier = Tier.T3,
validationReportId = ValidationReportId("vr-1"),
@@ -282,7 +305,8 @@ class DomainEventMapperTest {
sessionId = sessionId,
stageId = stageId,
projectId = null,
))
),
)
val result = domainEventToServerMessage(event, noopStore) as ServerMessage.ApprovalRequired
assertEquals(requestId, result.requestId)
assertEquals(Tier.T3, result.tier)
@@ -5,8 +5,8 @@ import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.EventMetadata
import com.correx.core.events.events.EventPayload
import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.stores.EventStore
@@ -27,6 +27,7 @@ class SessionEventBridgeTest {
private val sessionId = SessionId("session-1")
private val stageId = StageId("stage-1")
private val workflowId = "workflow-test"
private val timestamp = Instant.parse("2026-01-01T00:00:00Z")
private val noopArtifactStore: ArtifactStore = object : ArtifactStore {
@@ -65,8 +66,8 @@ class SessionEventBridgeTest {
@Test
fun `replaySnapshot sends mapped messages for all events`() = runTest {
val events = listOf(
storedEvent(WorkflowStartedEvent(sessionId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L),
)
val store = fakeEventStore(allEventsList = events)
val sent = mutableListOf<ServerMessage>()
@@ -75,7 +76,7 @@ class SessionEventBridgeTest {
bridge.replaySnapshot()
assertEquals(2, sent.size)
assertEquals(ServerMessage.SessionStarted(sessionId, stageId.value), sent[0])
assertEquals(ServerMessage.SessionStarted(sessionId, workflowId), sent[0])
assertEquals(ServerMessage.SessionCompleted(sessionId), sent[1])
}
@@ -83,7 +84,7 @@ class SessionEventBridgeTest {
fun `replaySnapshot skips unmapped events`() = runTest {
val events = listOf(
storedEvent(OrchestrationResumedEvent(sessionId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L),
)
val store = fakeEventStore(allEventsList = events)
val sent = mutableListOf<ServerMessage>()
@@ -98,8 +99,8 @@ class SessionEventBridgeTest {
@Test
fun `streamLive sends mapped messages from live flow`() = runTest {
val events = listOf(
storedEvent(WorkflowStartedEvent(sessionId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L),
storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L),
storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L),
)
val liveFlow: Flow<StoredEvent> = flow { events.forEach { emit(it) } }
val store = fakeEventStore(liveFlow = liveFlow)
@@ -109,7 +110,7 @@ class SessionEventBridgeTest {
bridge.streamLive(sessionId)
assertEquals(2, sent.size)
assertEquals(ServerMessage.SessionStarted(sessionId, stageId.value), sent[0])
assertEquals(ServerMessage.SessionStarted(sessionId, workflowId), sent[0])
assertEquals(ServerMessage.SessionCompleted(sessionId), sent[1])
}
@@ -123,10 +124,10 @@ class SessionEventBridgeTest {
val sent = mutableListOf<ServerMessage>()
val bridge = SessionEventBridge(store, noopArtifactStore) { sent.add(it) }
channel.send(storedEvent(WorkflowStartedEvent(sessionId, stageId), seq = 1L))
channel.send(storedEvent(WorkflowStartedEvent(sessionId, workflowId, stageId), seq = 1L))
val job = launch { bridge.streamLive(sessionId) }
channel.send(storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L))
channel.send(storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1, workflowId), seq = 2L))
channel.close()
job.join()
@@ -20,12 +20,12 @@ import dev.tamboui.terminal.Frame
import dev.tamboui.tui.EventHandler
import dev.tamboui.tui.Renderer
import dev.tamboui.tui.TuiRunner
import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent
private const val DEFAULT_PORT = 8080
private const val STATUS_HEIGHT = 1
@@ -70,7 +70,7 @@ fun main(args: Array<String>) {
val handler = EventHandler { event, _ ->
if (event is TambouiKeyEvent) {
mapKey(event, state.inputMode)
mapKey(event)
?.let { KeyResolver.resolve(it, state.inputMode, state.inputBuffer) }
?.let(::dispatch)
}
@@ -23,50 +23,11 @@ fun inputBarWidget(state: TuiState): Paragraph {
?: "(no session)"
val (row1, row2) = when (state.inputMode) {
InputMode.ROUTER -> {
val cursor = if (state.inputBuffer.isNotEmpty()) {
Line.from(Span.raw(""), Span.raw(state.inputBuffer))
} else {
Line.from(Span.raw(""), Span.styled("Ask anything…", dimStyle))
}
val hints = buildList<String> {
add("$sessionName · router")
add("tab navigate")
if (hasApproval) { add("ctrl+a approve"); add("ctrl+r reject") }
if (hasSession) add("ctrl+s steer")
add("ctrl+e events")
add("ctrl+q")
}.joinToString(" ")
cursor to Line.from(Span.styled(hints, dimStyle))
}
InputMode.NAVIGATE -> {
val row1Line = Line.from(Span.styled(" ↑↓ sessions enter select", dimStyle))
val hints = buildList<String> {
add("$sessionName · navigate")
add("tab router")
if (hasApproval) { add("ctrl+a approve"); add("ctrl+r reject") }
if (hasSession) add("ctrl+s steer")
add("ctrl+e events")
add("ctrl+q")
}.joinToString(" ")
row1Line to Line.from(Span.styled(hints, dimStyle))
}
InputMode.FILTER -> {
val cursor = if (state.inputBuffer.isNotEmpty()) {
Line.from(Span.raw(""), Span.raw(state.inputBuffer))
} else {
Line.from(Span.raw(""), Span.styled("/filter…", dimStyle))
}
cursor to Line.from(Span.styled("filtering sessions esc clear enter apply", dimStyle))
}
InputMode.STEER -> {
val cursor = if (state.inputBuffer.isNotEmpty()) {
Line.from(Span.raw(""), Span.raw(state.inputBuffer))
} else {
Line.from(Span.raw(""), Span.styled("Steer the session…", dimStyle))
}
cursor to Line.from(Span.styled("$sessionName · steering esc cancel enter send", dimStyle))
}
InputMode.ROUTER -> createRowsForRouter(state, dimStyle, sessionName, hasApproval, hasSession)
InputMode.NAVIGATE -> createRowsForNavigate(dimStyle, sessionName, hasApproval, hasSession)
InputMode.FILTER -> createRowsForFilter(state, dimStyle)
InputMode.STEER -> createRowsForSteer(state, dimStyle, sessionName)
}
val block = Block.builder()
@@ -77,3 +38,73 @@ fun inputBarWidget(state: TuiState): Paragraph {
return Paragraph.builder().text(Text.from(listOf(row1, row2))).block(block).build()
}
private fun createRowsForSteer(
state: TuiState,
dimStyle: Style?,
sessionName: String,
): Pair<Line?, Line?> {
val cursor = if (state.inputBuffer.isNotEmpty()) {
Line.from(Span.raw(""), Span.raw(state.inputBuffer))
} else {
Line.from(Span.raw(""), Span.styled("Steer the session…", dimStyle))
}
return cursor to Line.from(Span.styled("$sessionName · steering esc cancel enter send", dimStyle))
}
private fun createRowsForFilter(
state: TuiState,
dimStyle: Style?,
): Pair<Line?, Line?> {
val cursor = if (state.inputBuffer.isNotEmpty()) {
Line.from(Span.raw(""), Span.raw(state.inputBuffer))
} else {
Line.from(Span.raw(""), Span.styled("/filter…", dimStyle))
}
return cursor to Line.from(Span.styled("filtering sessions esc clear enter apply", dimStyle))
}
private fun createRowsForNavigate(
dimStyle: Style?,
sessionName: String,
hasApproval: Boolean,
hasSession: Boolean,
): Pair<Line?, Line?> {
val row1Line = Line.from(Span.styled(" ↑↓ sessions enter select", dimStyle))
val hints = buildList<String> {
add("$sessionName · navigate")
add("tab router")
if (hasApproval) {
add("ctrl+a approve"); add("ctrl+r reject")
}
if (hasSession) add("ctrl+s steer")
add("ctrl+e events")
add("ctrl+q")
}.joinToString(" ")
return row1Line to Line.from(Span.styled(hints, dimStyle))
}
private fun createRowsForRouter(
state: TuiState,
dimStyle: Style?,
sessionName: String,
hasApproval: Boolean,
hasSession: Boolean,
): Pair<Line?, Line?> {
val cursor = if (state.inputBuffer.isNotEmpty()) {
Line.from(Span.raw(""), Span.raw(state.inputBuffer))
} else {
Line.from(Span.raw(""), Span.styled("Ask anything…", dimStyle))
}
val hints = buildList<String> {
add("$sessionName · router")
add("tab navigate")
if (hasApproval) {
add("ctrl+a approve"); add("ctrl+r reject")
}
if (hasSession) add("ctrl+s steer")
add("ctrl+e events")
add("ctrl+q")
}.joinToString(" ")
return cursor to Line.from(Span.styled(hints, dimStyle))
}
@@ -1,7 +1,6 @@
package com.correx.apps.tui.components
import com.correx.apps.tui.state.TuiState
import dev.tamboui.style.Color
import dev.tamboui.style.Style
import dev.tamboui.text.Line
import dev.tamboui.text.Span
@@ -28,7 +28,6 @@ fun sessionListWidget(state: TuiState): Paragraph {
val isSelected = s.id == state.sessions.selectedId
val prefix = if (isSelected) Span.styled("", Style.create().blue()) else Span.raw(" ")
val idSpan = Span.styled("[${s.id.take(ID_LEN)}]", dimStyle)
val statusStr = s.status.uppercase().padEnd(STATUS_WIDTH)
val statusSpan = statusSpan(s)
val rawName = s.name.ifEmpty { s.workflowId }
val truncated = if (rawName.length > NAME_MAX) rawName.take(NAME_MAX - 1) + "" else rawName
@@ -61,6 +60,7 @@ private fun statusSpan(s: SessionSummary): Span {
fun filteredSessions(sessions: SessionsState): List<SessionSummary> {
val f = sessions.filter
return if (f.isEmpty()) sessions.sessions
else sessions.sessions.filter { it.workflowId.contains(f, ignoreCase = true) }
val sorted = sessions.sessions.sortedByDescending { it.lastEventAt }
return if (f.isEmpty()) sorted
else sorted.filter { it.workflowId.contains(f, ignoreCase = true) }
}
@@ -1,24 +1,23 @@
package com.correx.apps.tui.input
import com.correx.apps.tui.KeyEvent
import com.correx.apps.tui.state.InputMode
import dev.tamboui.tui.event.KeyCode
import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent
@Suppress("CyclomaticComplexMethod", "ReturnCount")
fun mapKey(event: TambouiKeyEvent, mode: InputMode): KeyEvent? {
if (event.isCtrlC()) return KeyEvent.Quit
fun mapKey(event: TambouiKeyEvent): KeyEvent? {
if (event.isCtrlC) return KeyEvent.Quit
if (event.hasAlt()) return null
if (event.hasCtrl()) {
return when (event.character()) {
'q' -> KeyEvent.Quit
'n' -> KeyEvent.NewSession
'c' -> KeyEvent.Cancel
'a' -> KeyEvent.Approve
'r' -> KeyEvent.Reject
's' -> KeyEvent.Steer
'h' -> KeyEvent.ToggleApprovalOverlay
'e' -> KeyEvent.ToggleEventOverlay
return when {
event.isChar('q') -> KeyEvent.Quit
event.isChar('n') -> KeyEvent.NewSession
event.isChar('c') -> KeyEvent.Cancel
event.isChar('a') -> KeyEvent.Approve
event.isChar('r') -> KeyEvent.Reject
event.isChar('s') -> KeyEvent.Steer
event.isChar('h') -> KeyEvent.ToggleApprovalOverlay
event.isChar('e') -> KeyEvent.ToggleEventOverlay
else -> null
}
}
@@ -17,11 +17,13 @@ object RootReducer {
val (approval, ae) = ApprovalReducer.reduce(afterInput.approval, prevInputMode, action)
val (connection, ce) = ConnectionReducer.reduce(afterInput.connection, action)
val (provider, pe) = ProviderReducer.reduce(afterInput.provider, action)
val approvalJustArrived = approval.active != null && afterInput.approval.active == null
val withSubReducers = afterInput.copy(
sessions = sessions,
approval = approval,
connection = connection,
provider = provider,
approvalOverlayVisible = if (approvalJustArrived) true else afterInput.approvalOverlayVisible,
)
val final = when (action) {
is Action.ToggleApprovalOverlay -> withSubReducers.copy(
@@ -25,8 +25,18 @@ object SessionsReducer {
action: Action,
clock: () -> Long = System::currentTimeMillis,
): Pair<SessionsState, List<Effect>> = when (action) {
is Action.NavigateUp -> if (inputMode == InputMode.NAVIGATE) navigateUp(sessions) to emptyList() else sessions to emptyList()
is Action.NavigateDown -> if (inputMode == InputMode.NAVIGATE) navigateDown(sessions) to emptyList() else sessions to emptyList()
is Action.NavigateUp -> if (inputMode == InputMode.NAVIGATE) {
navigateUp(sessions) to emptyList()
} else {
sessions to emptyList()
}
is Action.NavigateDown -> if (inputMode == InputMode.NAVIGATE) {
navigateDown(sessions) to emptyList()
} else {
sessions to emptyList()
}
is Action.SubmitInput -> if (inputMode == InputMode.FILTER) {
sessions.copy(filter = inputText) to emptyList()
} else {
@@ -52,8 +62,12 @@ object SessionsReducer {
else -> sessions to emptyList()
}
private fun filteredSessions(sessions: SessionsState): List<SessionSummary> =
if (sessions.filter.isBlank()) sessions.sessions
else sessions.sessions.filter { it.workflowId.contains(sessions.filter, ignoreCase = true) }
private fun navigateUp(sessions: SessionsState): SessionsState {
val list = sessions.sessions
val list = filteredSessions(sessions)
if (list.isEmpty()) return sessions
val idx = list.indexOfFirst { it.id == sessions.selectedId }
val newIdx = if (idx <= 0) list.lastIndex else idx - 1
@@ -61,7 +75,7 @@ object SessionsReducer {
}
private fun navigateDown(sessions: SessionsState): SessionsState {
val list = sessions.sessions
val list = filteredSessions(sessions)
if (list.isEmpty()) return sessions
val idx = list.indexOfFirst { it.id == sessions.selectedId }
val newIdx = if (idx >= list.lastIndex) 0 else idx + 1
@@ -77,58 +91,38 @@ object SessionsReducer {
msg: ServerMessage,
clock: () -> Long,
): Pair<SessionsState, List<Effect>> = when (msg) {
is ServerMessage.SessionStarted -> {
val summary = SessionSummary(
id = msg.sessionId.value,
status = "ACTIVE",
workflowId = msg.workflowId,
name = msg.workflowId,
lastEventAt = clock(),
currentStage = null,
lastOutput = null,
)
val selected = sessions.selectedId ?: msg.sessionId.value
sessions.copy(sessions = sessions.sessions + summary, selectedId = selected) to
listOf(Effect.ConnectSession(msg.sessionId))
is ServerMessage.SessionStarted -> processSessionStartedMessage(msg, clock, sessions)
is ServerMessage.SessionPaused -> processSessionPausedMessage(msg, sessions, clock)
is ServerMessage.SessionCompleted -> processSessionCompletedMessage(sessions, msg, clock)
is ServerMessage.SessionFailed -> processSessionFailedMessage(sessions, msg, clock)
is ServerMessage.StageStarted -> processStageStartedMessage(msg, sessions)
is ServerMessage.StageCompleted -> processStageCompletedMessage(msg, sessions)
is ServerMessage.StageFailed -> processStageFailedMessage(msg, sessions)
is ServerMessage.ToolStarted -> processToolStartedMessage(clock, sessions, msg)
is ServerMessage.InferenceCompleted -> processInferenceCompletedMessage(sessions, msg)
is ServerMessage.ToolCompleted -> processToolCompletedMessage(msg, sessions)
is ServerMessage.ToolFailed -> processToolFailedMessage(msg, sessions)
is ServerMessage.ToolRejected -> processToolRejectedMessage(clock, sessions, msg)
else -> sessions to emptyList()
}
is ServerMessage.SessionPaused -> {
val statusLabel = if (msg.reason == PauseReason.APPROVAL_PENDING) {
"PAUSED awaiting approval"
} else {
"PAUSED"
}
sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) s.copy(status = statusLabel, lastEventAt = clock()) else s
},
) to emptyList()
}
is ServerMessage.SessionCompleted -> touchSession(sessions, msg.sessionId.value, "COMPLETED", clock) to listOf(
Effect.DisconnectSession,
)
is ServerMessage.SessionFailed -> touchSession(
sessions,
msg.sessionId.value,
"FAILED",
clock,
) to listOf(Effect.DisconnectSession)
is ServerMessage.StageStarted -> {
private fun processToolRejectedMessage(
clock: () -> Long,
sessions: SessionsState,
msg: ServerMessage.ToolRejected,
): Pair<SessionsState, List<Effect>> {
val now = clock()
sessions.copy(
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val entry = TuiEventEntry(formatTime(now), "StageStarted", msg.stageId.value)
s.copy(
currentStage = msg.stageId.value,
currentStageId = msg.stageId.value,
tools = emptyList(),
recentEvents = (s.recentEvents + entry).takeLast(7),
lastEventAt = now,
)
val updatedTools = s.tools.map { t ->
if (t.name == msg.toolName && t.status == ToolDisplayStatus.STARTED) {
t.copy(status = ToolDisplayStatus.REJECTED)
} else {
t
}
}
s.copy(tools = updatedTools, lastEventAt = now)
} else {
s
}
@@ -136,17 +130,22 @@ object SessionsReducer {
) to emptyList()
}
is ServerMessage.StageCompleted -> {
val now = clock()
sessions.copy(
private fun processToolFailedMessage(
msg: ServerMessage.ToolFailed,
sessions: SessionsState,
): Pair<SessionsState, List<Effect>> {
val now = msg.occurredAt
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val entry = TuiEventEntry(formatTime(now), "StageCompleted", msg.stageId.value)
s.copy(
currentStage = null,
recentEvents = (s.recentEvents + entry).takeLast(7),
lastEventAt = now,
)
val updatedTools = s.tools.map { t ->
if (t.name == msg.toolName && t.status == ToolDisplayStatus.STARTED) {
t.copy(status = ToolDisplayStatus.FAILED)
} else {
t
}
}
s.copy(tools = updatedTools, lastEventAt = now)
} else {
s
}
@@ -154,45 +153,12 @@ object SessionsReducer {
) to emptyList()
}
is ServerMessage.StageFailed -> {
val now = clock()
sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val entry = TuiEventEntry(formatTime(now), "StageFailed", msg.stageId.value)
s.copy(
currentStage = null,
recentEvents = (s.recentEvents + entry).takeLast(7),
lastEventAt = now,
)
} else {
s
}
},
) to emptyList()
}
is ServerMessage.ToolStarted -> {
val now = clock()
sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val record = TuiToolRecord(msg.toolName, msg.tier.level, ToolDisplayStatus.STARTED, null)
s.copy(
tools = (s.tools + record).takeLast(8),
lastEventAt = now,
)
} else {
s
}
},
) to emptyList()
}
is ServerMessage.InferenceCompleted -> applyInferenceCompleted(sessions, msg, clock)
is ServerMessage.ToolCompleted -> {
val now = clock()
sessions.copy(
private fun processToolCompletedMessage(
msg: ServerMessage.ToolCompleted,
sessions: SessionsState,
): Pair<SessionsState, List<Effect>> {
val now = msg.occurredAt
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val updatedTools = s.tools.map { t ->
@@ -216,55 +182,153 @@ object SessionsReducer {
) to emptyList()
}
is ServerMessage.ToolFailed -> {
val now = clock()
sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val updatedTools = s.tools.map { t ->
if (t.name == msg.toolName && t.status == ToolDisplayStatus.STARTED) {
t.copy(status = ToolDisplayStatus.FAILED)
} else {
t
}
}
s.copy(tools = updatedTools, lastEventAt = now)
} else {
s
}
},
) to emptyList()
}
is ServerMessage.ToolRejected -> {
val now = clock()
sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val updatedTools = s.tools.map { t ->
if (t.name == msg.toolName && t.status == ToolDisplayStatus.STARTED) {
t.copy(status = ToolDisplayStatus.REJECTED)
} else {
t
}
}
s.copy(tools = updatedTools, lastEventAt = now)
} else {
s
}
},
) to emptyList()
}
else -> sessions to emptyList()
}
private fun applyInferenceCompleted(
sessions: SessionsState,
msg: ServerMessage.InferenceCompleted,
private fun processToolStartedMessage(
clock: () -> Long,
sessions: SessionsState,
msg: ServerMessage.ToolStarted,
): Pair<SessionsState, List<Effect>> {
val now = clock()
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val record = TuiToolRecord(msg.toolName, msg.tier.level, ToolDisplayStatus.STARTED, null)
s.copy(
tools = (s.tools + record).takeLast(8),
lastEventAt = now,
)
} else {
s
}
},
) to emptyList()
}
private fun processStageFailedMessage(
msg: ServerMessage.StageFailed,
sessions: SessionsState,
): Pair<SessionsState, List<Effect>> {
val now = msg.occurredAt
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val entry = TuiEventEntry(formatTime(now), "StageFailed", msg.stageId.value)
s.copy(
currentStage = null,
recentEvents = (s.recentEvents + entry).takeLast(7),
lastEventAt = now,
)
} else {
s
}
},
) to emptyList()
}
private fun processStageCompletedMessage(
msg: ServerMessage.StageCompleted,
sessions: SessionsState,
): Pair<SessionsState, List<Effect>> {
val now = msg.occurredAt
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val entry = TuiEventEntry(formatTime(now), "StageCompleted", msg.stageId.value)
s.copy(
currentStage = null,
recentEvents = (s.recentEvents + entry).takeLast(7),
lastEventAt = now,
)
} else {
s
}
},
) to emptyList()
}
private fun processStageStartedMessage(
msg: ServerMessage.StageStarted,
sessions: SessionsState,
): Pair<SessionsState, List<Effect>> {
val now = msg.occurredAt
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
val entry = TuiEventEntry(formatTime(now), "StageStarted", msg.stageId.value)
s.copy(
currentStage = msg.stageId.value,
currentStageId = msg.stageId.value,
tools = emptyList(),
recentEvents = (s.recentEvents + entry).takeLast(7),
lastEventAt = now,
)
} else {
s
}
},
) to emptyList()
}
private fun processSessionFailedMessage(
sessions: SessionsState,
msg: ServerMessage.SessionFailed,
clock: () -> Long,
): Pair<SessionsState, List<Effect.DisconnectSession>> = touchSession(
sessions,
msg.sessionId.value,
"FAILED",
clock,
) to listOf(Effect.DisconnectSession)
private fun processSessionCompletedMessage(
sessions: SessionsState,
msg: ServerMessage.SessionCompleted,
clock: () -> Long,
): Pair<SessionsState, List<Effect.DisconnectSession>> =
touchSession(sessions, msg.sessionId.value, "COMPLETED", clock) to listOf(
Effect.DisconnectSession,
)
private fun processSessionPausedMessage(
msg: ServerMessage.SessionPaused,
sessions: SessionsState,
clock: () -> Long,
): Pair<SessionsState, List<Effect>> {
val statusLabel = if (msg.reason == PauseReason.APPROVAL_PENDING) {
"PAUSED awaiting approval"
} else {
"PAUSED"
}
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) s.copy(status = statusLabel, lastEventAt = clock()) else s
},
) to emptyList()
}
private fun processSessionStartedMessage(
msg: ServerMessage.SessionStarted,
clock: () -> Long,
sessions: SessionsState,
): Pair<SessionsState, List<Effect.ConnectSession>> {
val summary = SessionSummary(
id = msg.sessionId.value,
status = "ACTIVE",
workflowId = msg.workflowId,
name = msg.workflowId,
lastEventAt = clock(),
currentStage = null,
lastOutput = null,
)
val selected = sessions.selectedId ?: msg.sessionId.value
return sessions.copy(sessions = sessions.sessions + summary, selectedId = selected) to
listOf(Effect.ConnectSession(msg.sessionId))
}
private fun processInferenceCompletedMessage(
sessions: SessionsState,
msg: ServerMessage.InferenceCompleted,
): Pair<SessionsState, List<Effect>> {
val now = msg.occurredAt
return sessions.copy(
sessions = sessions.sessions.map { s ->
if (s.id == msg.sessionId.value) {
@@ -1,13 +1,12 @@
package com.correx.apps.tui.input
import com.correx.apps.tui.KeyEvent
import com.correx.apps.tui.state.InputMode
import dev.tamboui.tui.event.KeyCode
import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent
import dev.tamboui.tui.event.KeyModifiers
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Test
import dev.tamboui.tui.event.KeyEvent as TambouiKeyEvent
class TambouiKeyMapperTest {
@@ -15,156 +14,156 @@ class TambouiKeyMapperTest {
@Test
fun `ctrl-c maps to Quit in InputMode ROUTER`() {
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code), InputMode.ROUTER))
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code)))
}
@Test
fun `ctrl-c maps to Quit in InputMode FILTER`() {
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code), InputMode.FILTER))
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code)))
}
@Test
fun `ctrl-c maps to Quit in InputMode NAVIGATE`() {
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code), InputMode.NAVIGATE))
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code)))
}
@Test
fun `ctrl-c maps to Quit in InputMode STEER`() {
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code), InputMode.STEER))
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'c'.code)))
}
// --- Ctrl+key → keybind events ---
@Test
fun `ctrl-q maps to Quit`() {
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'q'.code), InputMode.ROUTER))
assertEquals(KeyEvent.Quit, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'q'.code)))
}
@Test
fun `ctrl-a maps to Approve`() {
assertEquals(KeyEvent.Approve, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'a'.code), InputMode.ROUTER))
assertEquals(KeyEvent.Approve, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'a'.code)))
}
@Test
fun `ctrl-r maps to Reject`() {
assertEquals(KeyEvent.Reject, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'r'.code), InputMode.ROUTER))
assertEquals(KeyEvent.Reject, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'r'.code)))
}
@Test
fun `ctrl-n maps to NewSession`() {
assertEquals(KeyEvent.NewSession, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'n'.code), InputMode.ROUTER))
assertEquals(KeyEvent.NewSession, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'n'.code)))
}
@Test
fun `ctrl-s maps to Steer`() {
assertEquals(KeyEvent.Steer, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 's'.code), InputMode.ROUTER))
assertEquals(KeyEvent.Steer, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 's'.code)))
}
@Test
fun `ctrl-h maps to ToggleApprovalOverlay`() {
assertEquals(KeyEvent.ToggleApprovalOverlay, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'h'.code), InputMode.ROUTER))
assertEquals(KeyEvent.ToggleApprovalOverlay, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'h'.code)))
}
@Test
fun `ctrl-e maps to ToggleEventOverlay`() {
assertEquals(KeyEvent.ToggleEventOverlay, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'e'.code), InputMode.ROUTER))
assertEquals(KeyEvent.ToggleEventOverlay, mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'e'.code)))
}
@Test
fun `ctrl+unbound key maps to null`() {
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'z'.code), InputMode.ROUTER))
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.CTRL, 'z'.code)))
}
// --- Alt always → null ---
@Test
fun `alt modifier maps to null`() {
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.ALT, 'q'.code), InputMode.ROUTER))
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.ALT, 'q'.code)))
}
// --- Bare chars always → CharInput (never trigger keybinds) ---
@Test
fun `bare a maps to CharInput in ROUTER`() {
assertEquals(KeyEvent.CharInput('a'), mapKey(TambouiKeyEvent.ofChar('a'), InputMode.ROUTER))
assertEquals(KeyEvent.CharInput('a'), mapKey(TambouiKeyEvent.ofChar('a')))
}
@Test
fun `bare q maps to CharInput in ROUTER`() {
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q'), InputMode.ROUTER))
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q')))
}
@Test
fun `bare s maps to CharInput in ROUTER`() {
assertEquals(KeyEvent.CharInput('s'), mapKey(TambouiKeyEvent.ofChar('s'), InputMode.ROUTER))
assertEquals(KeyEvent.CharInput('s'), mapKey(TambouiKeyEvent.ofChar('s')))
}
@Test
fun `bare x maps to CharInput in ROUTER`() {
assertEquals(KeyEvent.CharInput('x'), mapKey(TambouiKeyEvent.ofChar('x'), InputMode.ROUTER))
assertEquals(KeyEvent.CharInput('x'), mapKey(TambouiKeyEvent.ofChar('x')))
}
@Test
fun `bare q maps to CharInput in FILTER`() {
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q'), InputMode.FILTER))
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q')))
}
@Test
fun `bare q maps to CharInput in STEER`() {
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q'), InputMode.STEER))
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q')))
}
@Test
fun `bare q maps to CharInput in NAVIGATE`() {
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q'), InputMode.NAVIGATE))
assertEquals(KeyEvent.CharInput('q'), mapKey(TambouiKeyEvent.ofChar('q')))
}
// --- KeyCode mappings ---
@Test
fun `UP maps to NavUp`() {
assertEquals(KeyEvent.NavUp, mapKey(TambouiKeyEvent.ofKey(KeyCode.UP), InputMode.ROUTER))
assertEquals(KeyEvent.NavUp, mapKey(TambouiKeyEvent.ofKey(KeyCode.UP)))
}
@Test
fun `DOWN maps to NavDown`() {
assertEquals(KeyEvent.NavDown, mapKey(TambouiKeyEvent.ofKey(KeyCode.DOWN), InputMode.ROUTER))
assertEquals(KeyEvent.NavDown, mapKey(TambouiKeyEvent.ofKey(KeyCode.DOWN)))
}
@Test
fun `ENTER maps to Enter`() {
assertEquals(KeyEvent.Enter, mapKey(TambouiKeyEvent.ofKey(KeyCode.ENTER), InputMode.ROUTER))
assertEquals(KeyEvent.Enter, mapKey(TambouiKeyEvent.ofKey(KeyCode.ENTER)))
}
@Test
fun `ESCAPE maps to Escape`() {
assertEquals(KeyEvent.Escape, mapKey(TambouiKeyEvent.ofKey(KeyCode.ESCAPE), InputMode.ROUTER))
assertEquals(KeyEvent.Escape, mapKey(TambouiKeyEvent.ofKey(KeyCode.ESCAPE)))
}
@Test
fun `BACKSPACE maps to Backspace`() {
assertEquals(KeyEvent.Backspace, mapKey(TambouiKeyEvent.ofKey(KeyCode.BACKSPACE), InputMode.ROUTER))
assertEquals(KeyEvent.Backspace, mapKey(TambouiKeyEvent.ofKey(KeyCode.BACKSPACE)))
}
@Test
fun `TAB maps to Filter`() {
assertEquals(KeyEvent.Filter, mapKey(TambouiKeyEvent.ofKey(KeyCode.TAB), InputMode.ROUTER))
assertEquals(KeyEvent.Filter, mapKey(TambouiKeyEvent.ofKey(KeyCode.TAB)))
}
@Test
fun `TAB maps to Filter in FILTER mode`() {
assertEquals(KeyEvent.Filter, mapKey(TambouiKeyEvent.ofKey(KeyCode.TAB), InputMode.FILTER))
assertEquals(KeyEvent.Filter, mapKey(TambouiKeyEvent.ofKey(KeyCode.TAB)))
}
// --- ISO control chars → null ---
@Test
fun `ESC codepoint as CHAR event maps to null`() {
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.NONE, 0x1B), InputMode.ROUTER))
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.NONE, 0x1B)))
}
@Test
fun `NUL codepoint as CHAR event maps to null`() {
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.NONE, 0x00), InputMode.ROUTER))
assertNull(mapKey(TambouiKeyEvent(KeyCode.CHAR, KeyModifiers.NONE, 0x00)))
}
}
@@ -51,7 +51,11 @@ class RootReducerTest {
)
val (state2, _) = RootReducer.reduce(state1, Action.ServerEventReceived(startMsg2), fixedClock)
// selected is s1 (first); navigating down should move to s2
val (state3, _) = RootReducer.reduce(state2.copy(inputMode = InputMode.NAVIGATE), Action.NavigateDown, fixedClock)
val (state3, _) = RootReducer.reduce(
state2.copy(inputMode = InputMode.NAVIGATE),
Action.NavigateDown,
fixedClock,
)
assertEquals("s2", state3.sessions.selectedId)
}
@@ -132,7 +132,11 @@ class SessionsReducerTest {
sessions = listOf(session("s1").copy(currentStage = "stage-1")),
selectedId = "s1",
)
val msg = ServerMessage.StageCompleted(sessionId = SessionId("s1"), stageId = StageId("stage-1"))
val msg = ServerMessage.StageCompleted(
sessionId = SessionId("s1"),
stageId = StageId("stage-1"),
occurredAt = fixedClock(),
)
val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg))
assertEquals(null, state.sessions[0].currentStage)
assertEquals(1000L, state.sessions[0].lastEventAt)
@@ -148,6 +152,7 @@ class SessionsReducerTest {
sessionId = SessionId("s1"),
stageId = StageId("stage-1"),
reason = "error",
occurredAt = fixedClock(),
)
val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg))
assertEquals(null, state.sessions[0].currentStage)
@@ -177,6 +182,7 @@ class SessionsReducerTest {
val msg = ServerMessage.InferenceCompleted(
sessionId = SessionId("s1"), stageId = StageId("stage-1"),
outputSummary = "summary", responseText = "full response",
occurredAt = fixedClock(),
)
val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg))
assertEquals("summary", state.sessions[0].lastOutput)
@@ -190,6 +196,7 @@ class SessionsReducerTest {
val msg = ServerMessage.InferenceCompleted(
sessionId = SessionId("s1"), stageId = StageId("stage-1"),
outputSummary = "", responseText = "",
occurredAt = fixedClock(),
)
val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg))
assertEquals(null, state.sessions[0].lastResponseText)
@@ -198,7 +205,11 @@ class SessionsReducerTest {
@Test
fun `ServerEventReceived StageStarted sets currentStage`() {
val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1")
val msg = ServerMessage.StageStarted(sessionId = SessionId("s1"), stageId = StageId("stage-1"))
val msg = ServerMessage.StageStarted(
sessionId = SessionId("s1"),
stageId = StageId("stage-1"),
occurredAt = fixedClock(),
)
val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg))
assertEquals("stage-1", state.sessions[0].currentStage)
assertEquals(1000L, state.sessions[0].lastEventAt)
@@ -207,7 +218,12 @@ class SessionsReducerTest {
@Test
fun `ServerEventReceived ToolCompleted updates lastOutput`() {
val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1")
val msg = ServerMessage.ToolCompleted(sessionId = SessionId("s1"), toolName = "file_write", outputSummary = "wrote 3 lines")
val msg = ServerMessage.ToolCompleted(
sessionId = SessionId("s1"),
toolName = "file_write",
outputSummary = "wrote 3 lines",
occurredAt = fixedClock(),
)
val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg))
assertEquals("file_write: wrote 3 lines", state.sessions[0].lastOutput)
assertEquals(1000L, state.sessions[0].lastEventAt)
@@ -8,6 +8,7 @@ import kotlinx.serialization.Serializable
@Serializable
data class WorkflowStartedEvent(
val sessionId: SessionId,
val workflowId: String,
val startStageId: StageId,
val retryPolicy: RetryPolicy? = null,
) : EventPayload
@@ -17,6 +18,7 @@ data class WorkflowCompletedEvent(
val sessionId: SessionId,
val terminalStageId: StageId,
val totalStages: Int,
val workflowId: String,
) : EventPayload
@Serializable
@@ -33,10 +33,10 @@ import com.correx.core.events.events.SessionFailedEvent
import com.correx.core.events.events.SessionPausedEvent
import com.correx.core.events.events.SessionResumedEvent
import com.correx.core.events.events.SessionStartedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.StageStartedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.ToolExecutionCompletedEvent
import com.correx.core.events.events.ToolExecutionFailedEvent
import com.correx.core.events.events.ToolExecutionRejectedEvent
@@ -1,14 +1,13 @@
package com.correx.core.kernel.orchestration
import com.correx.core.approvals.model.ApprovalDecision
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.artifacts.ArtifactState
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.ArtifactValidatedEvent
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.ArtifactLifecyclePhase
import org.slf4j.LoggerFactory
import com.correx.core.events.orchestration.OrchestrationState
import com.correx.core.events.types.ApprovalRequestId
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.kernel.execution.WorkflowResult
@@ -17,6 +16,7 @@ import com.correx.core.sessions.Session
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.transitions.graph.WorkflowGraph
import com.correx.core.transitions.resolution.TransitionDecision
import org.slf4j.LoggerFactory
import java.util.concurrent.*
import java.util.concurrent.atomic.*
@@ -94,7 +94,7 @@ class DefaultSessionOrchestrator(
}
is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) {
completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount)
completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount, enriched.graph.id)
} else {
failWorkflow(
sessionId = enriched.sessionId,
@@ -140,7 +140,7 @@ class DefaultSessionOrchestrator(
if (isTerminal(ctx.graph, nextStageId)) {
// Terminal stage is a sentinel — not executed, so don't count it.
return StepResult.Terminal(
completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount),
completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id),
)
}
@@ -150,6 +150,7 @@ class DefaultSessionOrchestrator(
currentStageId = advanceStage(ctx.sessionId, ctx.currentStageId, decision),
),
)
is StepResult.Terminal -> result
}
}
@@ -2,7 +2,6 @@ package com.correx.core.kernel.orchestration
import com.correx.core.approvals.domain.NoOpApprovalEngine
import com.correx.core.artifactstore.ArtifactStore
import org.slf4j.LoggerFactory
import com.correx.core.context.model.ContextPack
import com.correx.core.events.events.ArtifactValidatedEvent
import com.correx.core.events.events.ArtifactValidatingEvent
@@ -12,17 +11,18 @@ import com.correx.core.events.types.StageId
import com.correx.core.inference.GenerationConfig
import com.correx.core.inference.InferenceRequest
import com.correx.core.inference.ResponseFormat
import com.correx.core.sessions.Session
import com.correx.core.kernel.execution.ReplayStrategy
import com.correx.core.kernel.execution.WorkflowResult
import com.correx.core.kernel.replay.ReplayInferenceProvider
import com.correx.core.kernel.retry.exception.ReplayArtifactMissingException
import com.correx.core.risk.NoOpRiskAssessor
import com.correx.core.sessions.Session
import com.correx.core.transitions.execution.StageExecutionResult
import com.correx.core.transitions.graph.StageConfig
import com.correx.core.transitions.graph.WorkflowGraph
import com.correx.core.transitions.resolution.TransitionDecision
import com.correx.core.validation.model.ValidationContext
import org.slf4j.LoggerFactory
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.*
@@ -87,7 +87,7 @@ class ReplayOrchestrator(
val nextStageId = decision.to
if (isTerminal(ctx.graph, nextStageId)) {
return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount)
return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount, ctx.graph.id)
}
when (val result = enterStage(ctx, nextStageId, session)) {
@@ -101,7 +101,7 @@ class ReplayOrchestrator(
}
is TransitionDecision.Stay -> if (isTerminal(ctx.graph, ctx.currentStageId)) {
completeWorkflow(ctx.sessionId, ctx.currentStageId, ctx.stageCount)
completeWorkflow(ctx.sessionId, ctx.currentStageId, ctx.stageCount, ctx.graph.id)
} else {
step(ctx)
}
@@ -2,22 +2,24 @@ package com.correx.core.kernel.orchestration
import com.correx.core.approvals.ApprovalOutcome
import com.correx.core.approvals.ApprovalStatus
import com.correx.core.approvals.Tier
import com.correx.core.approvals.model.ApprovalDecision
import com.correx.core.approvals.model.DomainApprovalRequest
import com.correx.core.artifacts.ArtifactState
import com.correx.core.artifacts.kind.JsonSchema
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.context.builder.ContextPackBuilder
import com.correx.core.context.model.ContextEntry
import com.correx.core.context.model.ContextLayer
import com.correx.core.context.model.ContextPack
import com.correx.core.context.model.TokenBudget
import com.correx.core.events.types.ContextEntryId
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.ArtifactCreatedEvent
import com.correx.core.events.events.ArtifactValidatedEvent
import com.correx.core.events.events.ArtifactValidatingEvent
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.EventMetadata
import com.correx.core.events.events.EventPayload
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.InferenceCompletedEvent
import com.correx.core.events.events.InferenceFailedEvent
import com.correx.core.events.events.InferenceStartedEvent
@@ -26,27 +28,25 @@ import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.RiskAssessedEvent
import com.correx.core.events.events.ToolInvocationRequestedEvent
import com.correx.core.events.events.ToolRequest
import com.correx.core.events.events.TransitionExecutedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.artifacts.ArtifactState
import com.correx.core.artifacts.kind.JsonSchema
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.ApprovalDecisionId
import com.correx.core.events.types.ApprovalRequestId
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.ContextEntryId
import com.correx.core.events.types.ContextPackId
import com.correx.core.events.types.EventId
import com.correx.core.events.types.InferenceRequestId
import com.correx.core.events.types.RiskSummaryId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.events.types.ValidationReportId
import com.correx.core.approvals.Tier
import com.correx.core.events.events.ToolInvocationRequestedEvent
import com.correx.core.events.events.ToolRequest
import com.correx.core.events.types.ToolInvocationId
import com.correx.core.events.types.ValidationReportId
import com.correx.core.inference.FinishReason
import com.correx.core.inference.InferenceRepository
import com.correx.core.inference.InferenceRequest
@@ -56,14 +56,14 @@ import com.correx.core.inference.ResponseFormat
import com.correx.core.inference.ToolCallRequest
import com.correx.core.inference.ToolDefinition
import com.correx.core.inference.ToolFunction
import com.correx.core.tools.contract.ToolExecutor
import com.correx.core.tools.contract.ToolResult
import com.correx.core.tools.registry.ToolRegistry
import com.correx.core.kernel.execution.WorkflowResult
import com.correx.core.risk.RiskAssessor
import com.correx.core.risk.RiskContext
import com.correx.core.risk.toApprovalTier
import com.correx.core.sessions.Session
import com.correx.core.tools.contract.ToolExecutor
import com.correx.core.tools.contract.ToolResult
import com.correx.core.tools.registry.ToolRegistry
import com.correx.core.transitions.evaluation.EvaluationContext
import com.correx.core.transitions.evaluation.PromptResolver
import com.correx.core.transitions.execution.StageExecutionResult
@@ -587,6 +587,7 @@ abstract class SessionOrchestrator(
sessionId,
WorkflowStartedEvent(
sessionId = sessionId,
workflowId = graph.id,
startStageId = graph.start,
retryPolicy = config.retryPolicy,
),
@@ -597,12 +598,13 @@ abstract class SessionOrchestrator(
sessionId: SessionId,
terminalStageId: StageId,
stageCount: Int,
workflowId: String,
): WorkflowResult.Completed {
log.debug(
"[Orchestrator] COMPLETED session={} terminalStage={} stages={}",
sessionId.value, terminalStageId.value, stageCount,
)
emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount))
emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount, workflowId))
cancellations.remove(sessionId)
return WorkflowResult.Completed(sessionId, terminalStageId)
}
@@ -13,7 +13,7 @@ import com.correx.core.router.model.RouterConfig
import com.correx.core.router.model.RouterL2Entry
import com.correx.core.router.model.RouterState
import com.correx.core.router.model.RouterTurn
import java.util.UUID
import java.util.*
interface RouterContextBuilder {
fun build(state: RouterState, budget: TokenBudget): ContextPack
@@ -5,11 +5,11 @@ import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.EventId
import com.correx.core.events.types.InferenceRequestId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.inference.GenerationConfig
import com.correx.core.inference.InferenceRequest
import com.correx.core.events.types.InferenceRequestId
import com.correx.core.inference.InferenceRouter
import com.correx.core.inference.ModelCapability
import com.correx.core.inference.ResponseFormat
@@ -18,8 +18,8 @@ import com.correx.core.router.model.RouterResponse
import com.correx.core.router.model.RouterTurn
import com.correx.core.router.model.TurnRole
import kotlinx.datetime.Clock
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import java.util.*
import java.util.concurrent.*
interface RouterFacade {
suspend fun onUserInput(
@@ -2,9 +2,9 @@ package com.correx.core.router
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
@@ -10,6 +10,7 @@ data class WorkflowGraph(
) {
val stageIds: Set<StageId> get() = stages.keys
init {
require(id.isNotBlank()) { "workflow id must not be blank" }
require(start in stages) { "start stage must exist in stages" }
}
}
@@ -1,9 +1,9 @@
package com.correx.core.validation.artifact
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.artifacts.kind.FileWrittenArtifact
import com.correx.core.artifacts.kind.ProcessResultArtifact
import com.correx.core.artifacts.kind.TypedArtifactSlot
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.types.ArtifactId
import com.correx.core.validation.model.ValidationContext
import com.correx.core.validation.model.ValidationIssue
+1 -1
View File
@@ -381,7 +381,7 @@ end-to-end example.
### Skeleton
```kotlin
fun main() = runBlocking {
fun main(): Unit = runBlocking {
var state = TuiState()
val ws = TuiWsClient(host = "localhost", port = 8080)
val effectScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
@@ -8,8 +8,8 @@ import com.correx.infrastructure.artifactscas.compact.CompactionReport
import com.correx.infrastructure.artifactscas.compact.Compactor
import com.correx.infrastructure.artifactscas.compact.LivenessScanner
import com.correx.infrastructure.artifactscas.config.CasConfig
import com.correx.infrastructure.artifactscas.evict.Evictor
import com.correx.infrastructure.artifactscas.evict.EvictionReport
import com.correx.infrastructure.artifactscas.evict.Evictor
import com.correx.infrastructure.artifactscas.index.ArtifactIndex
import com.correx.infrastructure.artifactscas.recovery.TailScanReport
import com.correx.infrastructure.artifactscas.recovery.TailScanner
@@ -3,7 +3,7 @@ package com.correx.infrastructure.artifactscas.segment
import org.bouncycastle.crypto.digests.Blake3Digest
import java.nio.ByteBuffer
import java.nio.ByteOrder
import java.util.zip.CRC32C
import java.util.zip.*
object SegmentLayout {
const val HASH_SIZE: Int = 32
@@ -24,7 +24,7 @@ import org.junit.jupiter.api.Test
import org.junit.jupiter.api.io.TempDir
import java.nio.file.Files
import java.nio.file.Path
import java.util.UUID
import java.util.*
import kotlin.io.path.listDirectoryEntries
class CompactorTest {
@@ -21,7 +21,7 @@ import org.junit.jupiter.api.Test
import org.junit.jupiter.api.io.TempDir
import java.nio.file.Files
import java.nio.file.Path
import java.util.UUID
import java.util.*
import kotlin.io.path.listDirectoryEntries
class EvictorTest {
@@ -19,7 +19,7 @@ import kotlinx.coroutines.withContext
import kotlinx.datetime.Instant
import java.sql.Connection
import java.sql.ResultSet
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.*
class SqliteEventStore(
private val connection: Connection,
@@ -21,7 +21,7 @@ import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.launch
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.*
class LiveArtifactRepository(
private val eventStore: EventStore,
@@ -1,5 +1,7 @@
package com.correx.infrastructure
import com.correx.core.artifacts.DefaultArtifactReducer
import com.correx.core.artifacts.repository.ArtifactRepository
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.EventDispatcher
import com.correx.core.events.stores.EventStore
@@ -8,8 +10,19 @@ import com.correx.core.inference.InferenceProvider
import com.correx.core.inference.InferenceRouter
import com.correx.core.inference.ProviderRegistry
import com.correx.core.inference.RoutingStrategy
import com.correx.core.router.DefaultRouterContextBuilder
import com.correx.core.router.DefaultRouterFacade
import com.correx.core.router.DefaultRouterReducer
import com.correx.core.router.DefaultRouterRepository
import com.correx.core.router.RouterFacade
import com.correx.core.router.RouterProjector
import com.correx.core.router.model.RouterConfig
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
import com.correx.core.tools.contract.ToolExecutor
import com.correx.core.tools.registry.ToolRegistry
import com.correx.infrastructure.artifactscas.CasArtifactStore
import com.correx.infrastructure.artifactscas.config.CasConfig
import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex
import com.correx.infrastructure.inference.DefaultProviderRegistry
import com.correx.infrastructure.inference.FirstAvailableRoutingStrategy
import com.correx.infrastructure.inference.commons.ModelDescriptor
@@ -17,11 +30,8 @@ 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.artifactscas.CasArtifactStore
import kotlinx.coroutines.runBlocking
import com.correx.infrastructure.artifactscas.config.CasConfig
import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex
import com.correx.infrastructure.persistence.SqliteEventStore
import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository
import com.correx.infrastructure.tools.DefaultToolRegistry
import com.correx.infrastructure.tools.DispatchingToolExecutor
import com.correx.infrastructure.tools.SandboxedToolExecutor
@@ -31,18 +41,8 @@ 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 com.correx.core.router.DefaultRouterContextBuilder
import com.correx.core.router.DefaultRouterFacade
import com.correx.core.router.DefaultRouterRepository
import com.correx.core.router.DefaultRouterReducer
import com.correx.core.router.RouterFacade
import com.correx.core.router.RouterProjector
import com.correx.core.router.model.RouterConfig
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
import io.ktor.client.HttpClient
import kotlinx.coroutines.runBlocking
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.Paths
@@ -32,7 +32,7 @@ class FileReadTool(
(request.parameters["path"] as? String)?.let { pathString ->
runCatching {
val path = Paths.get(pathString).normalize().toAbsolutePath()
if (allowedPaths.isNotEmpty() && allowedPaths.none { path.startsWith(it.normalize().toAbsolutePath()) }) {
if (allowedPaths.none { path.startsWith(it.normalize().toAbsolutePath()) }) {
return@runCatching ValidationResult.Invalid("Path '$pathString' is not in the allowed list")
}
ValidationResult.Valid
@@ -65,7 +65,11 @@ class FileWriteTool(
}
put(
"required",
buildJsonArray { add(JsonPrimitive("path")); add(JsonPrimitive("content")); add(JsonPrimitive("operation")) },
buildJsonArray {
add(JsonPrimitive("path"))
add(JsonPrimitive("content"))
add(JsonPrimitive("operation"))
},
)
}
override val tier: Tier = Tier.T2
@@ -33,7 +33,7 @@ class FileEditToolTest {
}
@Test
fun `execute patch success`() = runBlocking {
fun `execute patch success`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_patch")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "line1\nline2\nline3\n")
@@ -63,7 +63,7 @@ class FileEditToolTest {
}
@Test
fun `validateRequest returns Invalid for disallowed path`() = runBlocking {
fun `validateRequest returns Invalid for disallowed path`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_test")
val otherDir = Files.createTempDirectory("other_dir")
val tool = FileEditTool(allowedPaths = setOf(tempDir))
@@ -80,7 +80,7 @@ class FileEditToolTest {
}
@Test
fun `validateRequest returns Invalid for non-existent file`() = runBlocking {
fun `validateRequest returns Invalid for non-existent file`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_test")
val tool = FileEditTool(allowedPaths = setOf(tempDir))
val request = createRequest(
@@ -96,7 +96,7 @@ class FileEditToolTest {
}
@Test
fun `validateRequest returns Invalid for missing parameters`() = runBlocking {
fun `validateRequest returns Invalid for missing parameters`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_test")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "content")
@@ -120,7 +120,7 @@ class FileEditToolTest {
}
@Test
fun `execute append success`() = runBlocking {
fun `execute append success`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_append")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "hello")
@@ -139,7 +139,7 @@ class FileEditToolTest {
}
@Test
fun `execute replace success`() = runBlocking {
fun `execute replace success`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_replace")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "the quick brown fox")
@@ -159,7 +159,7 @@ class FileEditToolTest {
}
@Test
fun `execute replace failure zero matches`() = runBlocking {
fun `execute replace failure zero matches`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_replace_fail")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "the quick brown fox")
@@ -180,7 +180,7 @@ class FileEditToolTest {
}
@Test
fun `execute replace failure multiple matches`() = runBlocking {
fun `execute replace failure multiple matches`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_replace_fail_multi")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "fox fox fox")
@@ -202,7 +202,7 @@ class FileEditToolTest {
@Test
fun `execute patch failure`() = runBlocking {
fun `execute patch failure`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_edit_patch_fail")
val filePath = tempDir.resolve("test.txt")
Files.writeString(filePath, "original")
@@ -31,15 +31,15 @@ class FileReadToolTest {
}
@Test
fun `validateRequest returns Valid for allowed path`() = runBlocking {
val tool = FileReadTool(allowedPaths = emptySet())
fun `validateRequest returns Valid for allowed path`(): Unit = runBlocking {
val tempFile = Files.createTempFile("test_allowed", ".txt")
val tool = FileReadTool(allowedPaths = setOf(tempFile.parent))
val request = createRequest(tempFile.toString())
assertEquals(ValidationResult.Valid, tool.validateRequest(request))
}
@Test
fun `validateRequest returns Invalid for disallowed path`() = runBlocking {
fun `validateRequest returns Invalid for disallowed path`(): Unit = runBlocking {
val tempFile = Files.createTempFile("test_disallowed", ".txt")
val tool = FileReadTool(allowedPaths = setOf(tempFile.parent.resolve("subdir")))
val request = createRequest(tempFile.toString())
@@ -49,7 +49,7 @@ class FileReadToolTest {
}
@Test
fun `validateRequest returns Invalid for missing path parameter`() = runBlocking {
fun `validateRequest returns Invalid for missing path parameter`(): Unit = runBlocking {
val tool = FileReadTool()
val request = ToolRequest(
invocationId = invocationId,
@@ -67,12 +67,12 @@ class FileReadToolTest {
}
@Test
fun `execute returns Success for valid file`() = runBlocking {
fun `execute returns Success for valid file`(): Unit = runBlocking {
val content = "Hello, world!"
val tempFile = Files.createTempFile("test_content", ".txt")
Files.writeString(tempFile, content)
val tool = FileReadTool(allowedPaths = emptySet())
val tool = FileReadTool(allowedPaths = setOf(tempFile.parent))
val request = createRequest(tempFile.toString())
val result = tool.execute(request)
@@ -83,7 +83,7 @@ class FileReadToolTest {
}
@Test
fun `execute returns Failure for non-existent file`() = runBlocking {
fun `execute returns Failure for non-existent file`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("test_dir")
val nonExistentFile = tempDir.resolve("missing.txt")
@@ -98,7 +98,7 @@ class FileReadToolTest {
}
@Test
fun `execute returns Failure for disallowed path`() = runBlocking {
fun `execute returns Failure for disallowed path`(): Unit = runBlocking {
val tempFile = Files.createTempFile("test_disallowed_exec", ".txt")
val tool = FileReadTool(allowedPaths = setOf(tempFile.parent.resolve("other")))
val request = createRequest(tempFile.toString())
@@ -34,7 +34,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Valid for allowed path and valid operation`() = runBlocking {
fun `validateRequest returns Valid for allowed path and valid operation`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val request = createRequest(
@@ -48,7 +48,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Invalid for disallowed path`() = runBlocking {
fun `validateRequest returns Invalid for disallowed path`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val otherDir = Files.createTempDirectory("other_dir")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
@@ -65,7 +65,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Invalid for missing operation`() = runBlocking {
fun `validateRequest returns Invalid for missing operation`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val request = createRequest(
@@ -83,7 +83,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Invalid for missing path`() = runBlocking {
fun `validateRequest returns Invalid for missing path`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val request = createRequest(
@@ -98,7 +98,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Invalid for missing content on write`() = runBlocking {
fun `validateRequest returns Invalid for missing content on write`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val request = createRequest(
@@ -113,7 +113,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Invalid for unknown operation`() = runBlocking {
fun `validateRequest returns Invalid for unknown operation`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val request = createRequest(
@@ -128,7 +128,7 @@ class FileWriteToolTest {
}
@Test
fun `execute returns Success for write operation`() = runBlocking {
fun `execute returns Success for write operation`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val filePath = tempDir.resolve("test.txt")
@@ -151,7 +151,7 @@ class FileWriteToolTest {
}
@Test
fun `execute returns Success for delete operation`() = runBlocking {
fun `execute returns Success for delete operation`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val filePath = tempDir.resolve("to_delete.txt")
@@ -164,12 +164,7 @@ class FileWriteToolTest {
),
)
println("TempDir: $tempDir")
println("FilePath: $filePath")
println("PathString from request: ${request.parameters["path"]}")
val result = tool.execute(request)
println("Result: $result")
println("Exists after: ${Files.exists(filePath)}")
assertTrue(result is ToolResult.Success)
val success = result as ToolResult.Success
assertEquals("File deleted successfully: $filePath", success.output)
@@ -178,7 +173,7 @@ class FileWriteToolTest {
}
@Test
fun `execute returns Failure for deleting non-existent file`() = runBlocking {
fun `execute returns Failure for deleting non-existent file`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
val filePath = tempDir.resolve("non_existent.txt")
@@ -199,7 +194,7 @@ class FileWriteToolTest {
}
@Test
fun `execute returns Failure for disallowed path`() = runBlocking {
fun `execute returns Failure for disallowed path`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test")
val otherDir = Files.createTempDirectory("other_dir")
val tool = FileWriteTool(allowedPaths = setOf(tempDir))
@@ -221,7 +216,7 @@ class FileWriteToolTest {
}
@Test
fun `validateRequest returns Valid for existing file in allowed directory`() = runBlocking {
fun `validateRequest returns Valid for existing file in allowed directory`(): Unit = runBlocking {
val tempDir = Files.createTempDirectory("file_write_test_existing")
val filePath = tempDir.resolve("existing.txt")
Files.writeString(filePath, "content")
@@ -21,7 +21,6 @@ import java.io.IOException
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.StandardCopyOption
import java.util.*
@SuppressWarnings("TooManyFunctions", "LongParameterList")
class SandboxedToolExecutor(
@@ -1,7 +1,7 @@
package com.correx.infrastructure.tools
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.artifacts.MaterializingArtifactWriter
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.tools.contract.Tool
import com.correx.infrastructure.tools.filesystem.FileEditTool
import com.correx.infrastructure.tools.filesystem.FileReadTool
@@ -25,41 +25,55 @@ data class FileWriteConfig(
val sandboxRoot: Path? = null,
val workingDir: Path? = null,
)
data class FileEditConfig(
val enabled: Boolean = false,
val allowedPaths: Set<Path> = emptySet(),
val workingDir: Path? = null,
)
data class ShellConfig(val enabled: Boolean = false, val allowedExecutables: Set<String> = emptySet(), val workingDir: Path? = null)
data class ShellConfig(
val enabled: Boolean = false,
val allowedExecutables: Set<String> = emptySet(),
val workingDir: Path? = null,
)
/**
* Extension function to convert [ToolConfig] into a list of [Tool] implementations.
*/
fun ToolConfig.buildTools(): List<Tool> = buildList {
if (shell.enabled) {
add(ShellTool(
add(
ShellTool(
allowedExecutables = shell.allowedExecutables,
workingDir = shell.workingDir,
))
),
)
}
if (fileRead.enabled) {
add(FileReadTool(
allowedPaths = fileRead.allowedPaths
))
add(
FileReadTool(
allowedPaths = fileRead.allowedPaths,
),
)
}
if (fileWrite.enabled) {
add(FileWriteTool(
add(
FileWriteTool(
allowedPaths = fileWrite.allowedPaths,
artifactStore = fileWrite.artifactStore,
materializingWriter = fileWrite.materializingWriter,
sandboxRoot = fileWrite.sandboxRoot,
workingDir = fileWrite.workingDir,
))
),
)
}
if (fileEdit.enabled) {
add(FileEditTool(
add(
FileEditTool(
allowedPaths = fileEdit.allowedPaths,
workingDir = fileEdit.workingDir,
))
),
)
}
}
@@ -30,14 +30,14 @@ class ShellToolTest {
}
@Test
fun `validateRequest returns Valid for allowed executable`() = runBlocking {
fun `validateRequest returns Valid for allowed executable`(): Unit = runBlocking {
val tool = ShellTool(allowedExecutables = setOf("echo"))
val request = createRequest(listOf("echo", "hello"))
assertEquals(ValidationResult.Valid, tool.validateRequest(request))
}
@Test
fun `validateRequest returns Invalid for disallowed executable`() = runBlocking {
fun `validateRequest returns Invalid for disallowed executable`(): Unit = runBlocking {
val tool = ShellTool(allowedExecutables = setOf("echo"))
val request = createRequest(listOf("ls"))
val result = tool.validateRequest(request)
@@ -46,7 +46,7 @@ class ShellToolTest {
}
@Test
fun `validateRequest returns Invalid for empty argv`() = runBlocking {
fun `validateRequest returns Invalid for empty argv`(): Unit = runBlocking {
val tool = ShellTool(allowedExecutables = setOf("echo"))
val request = createRequest(emptyList())
val result = tool.validateRequest(request)
@@ -58,7 +58,7 @@ class ShellToolTest {
}
@Test
fun `validateRequest returns Invalid for null argv`() = runBlocking {
fun `validateRequest returns Invalid for null argv`(): Unit = runBlocking {
val tool = ShellTool(allowedExecutables = setOf("echo"))
val request = ToolRequest(
invocationId = invocationId,
@@ -76,7 +76,7 @@ class ShellToolTest {
}
@Test
fun `execute returns Success for exit code 0`() = runBlocking {
fun `execute returns Success for exit code 0`(): Unit = runBlocking {
val tool = ShellTool(allowedExecutables = setOf("echo"))
val request = createRequest(listOf("echo", "hello"))
val result = tool.execute(request)
@@ -89,7 +89,7 @@ class ShellToolTest {
}
@Test
fun `execute returns Failure with non-zero exit code for failed process`() = runBlocking {
fun `execute returns Failure with non-zero exit code for failed process`(): Unit = runBlocking {
val tool = ShellTool(allowedExecutables = setOf("false"))
val request = createRequest(listOf("false"))
val result = tool.execute(request)
@@ -101,7 +101,7 @@ class ShellToolTest {
}
@Test
fun `execute returns Failure on timeout`() = runBlocking {
fun `execute returns Failure on timeout`(): Unit = runBlocking {
// Using sleep command to simulate timeout
val tool = ShellTool(allowedExecutables = setOf("sleep"), timeoutMs = 100L)
val request = createRequest(listOf("sleep", "1"))
@@ -32,12 +32,12 @@ private data class ProducesEntry(
private data class StageSection(
val id: String = "",
val prompt: String? = null,
@JsonProperty("system_prompt") val systemPrompt: String? = null,
@param:JsonProperty("system_prompt") val systemPrompt: String? = null,
val produces: List<ProducesEntry> = emptyList(),
val needs: List<String> = emptyList(),
@JsonProperty("allowed_tools") val allowedTools: List<String> = emptyList(),
@JsonProperty("token_budget") val tokenBudget: Int = 4096,
@JsonProperty("max_retries") val maxRetries: Int = 3,
@param:JsonProperty("allowed_tools") val allowedTools: List<String> = emptyList(),
@param:JsonProperty("token_budget") val tokenBudget: Int = 4096,
@param:JsonProperty("max_retries") val maxRetries: Int = 3,
)
// condition fields flattened into the transition row
@@ -45,10 +45,10 @@ 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,
@param:JsonProperty("condition_type") val conditionType: String = "",
@param:JsonProperty("condition_artifact_id") val conditionArtifactId: String? = null,
@param:JsonProperty("condition_key") val conditionKey: String? = null,
@param:JsonProperty("condition_value") val conditionValue: String? = null,
)
private const val TERMINAL = "done"
@@ -4,8 +4,8 @@ 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
import kotlin.test.assertEquals
class FileSystemPromptLoaderTest {
@@ -17,6 +17,7 @@ class WorkflowGraphTest {
@Test
fun `graph stores immutable structure`() {
val graph = WorkflowGraph(
id = "workflow-test",
stages = mapOf(s1 to StageConfig(), s2 to StageConfig()),
transitions = setOf(
TransitionEdge(
@@ -39,6 +40,7 @@ class WorkflowGraphTest {
val node = s1
val graph = WorkflowGraph(
id = "workflow-test",
stages = mapOf(node to StageConfig()),
transitions = emptySet(),
start = s1
@@ -51,12 +53,14 @@ class WorkflowGraphTest {
@Test
fun `graph equality is structural`() {
val g1 = WorkflowGraph(
id = "workflow-test",
stages = mapOf(s1 to StageConfig()),
transitions = emptySet(),
start = s1
)
val g2 = WorkflowGraph(
id = "workflow-test",
stages = mapOf(s1 to StageConfig()),
transitions = emptySet(),
start = s1
@@ -14,11 +14,15 @@ import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Test
class EventsTest {
private val workflowId = "workflow-test"
@Test
fun `WorkflowStartedEvent survives round-trip`() {
val event: EventPayload = WorkflowStartedEvent(
sessionId = SessionId("s1"),
startStageId = StageId("stage-a"),
workflowId = workflowId,
)
val json = eventJson.encodeToString(EventPayload.serializer(), event)
@@ -33,6 +37,7 @@ class EventsTest {
sessionId = SessionId("s1"),
terminalStageId = StageId("stage-a"),
totalStages = 1,
workflowId = workflowId,
)
val json = eventJson.encodeToString(EventPayload.serializer(), event)
@@ -103,6 +103,7 @@ class GraphValidatorTest {
)
val graph = WorkflowGraph(
id = "workflow-test",
stages = mapOf(a to StageConfig(), b to StageConfig()),
transitions = setOf(edge1, edge2),
start = a
@@ -1,6 +1,7 @@
import com.correx.core.context.model.CompressionMetadata
import com.correx.core.context.model.ContextLayer
import com.correx.core.context.model.TokenBudget
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.router.DefaultRouterContextBuilder
import com.correx.core.router.model.RouterConfig
import com.correx.core.router.model.RouterL2Entry
@@ -9,10 +10,6 @@ import com.correx.core.router.model.RouterTurn
import com.correx.core.router.model.StageOutcomeKind
import com.correx.core.router.model.TurnRole
import com.correx.core.router.model.WorkflowStatus
import com.correx.core.events.types.ContextEntryId
import com.correx.core.events.types.ContextPackId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import kotlinx.datetime.Clock
import kotlinx.datetime.Instant
import org.junit.jupiter.api.Assertions.assertEquals
@@ -537,8 +534,18 @@ class RouterContextBuilderTest {
RouterTurn(TurnRole.ROUTER, "second reply", Instant.parse("2026-01-04T00:00:00Z")),
),
l2Memory = listOf(
RouterL2Entry(StageId("s1"), "stage one done", StageOutcomeKind.SUCCESS, Instant.parse("2026-01-01T10:00:00Z")),
RouterL2Entry(StageId("s2"), "stage two failed", StageOutcomeKind.FAILURE, Instant.parse("2026-01-02T10:00:00Z")),
RouterL2Entry(
StageId("s1"),
"stage one done",
StageOutcomeKind.SUCCESS,
Instant.parse("2026-01-01T10:00:00Z"),
),
RouterL2Entry(
StageId("s2"),
"stage two failed",
StageOutcomeKind.FAILURE,
Instant.parse("2026-01-02T10:00:00Z"),
),
),
)
val pack = builder.build(state, TokenBudget(limit = 10000))
@@ -1,12 +1,13 @@
import com.correx.core.context.model.ContextPack
import com.correx.core.context.model.TokenBudget
import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.StoredEvent
import com.correx.core.events.stores.EventStore
import com.correx.core.events.types.ContextPackId
import com.correx.core.events.types.InferenceRequestId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.inference.GenerationConfig
import com.correx.core.inference.InferenceProvider
import com.correx.core.inference.InferenceRequest
import com.correx.core.inference.InferenceResponse
@@ -14,7 +15,6 @@ import com.correx.core.inference.InferenceRouter
import com.correx.core.inference.ModelCapability
import com.correx.core.inference.TokenUsage
import com.correx.core.router.ChatMode
import com.correx.core.router.DefaultRouterContextBuilder
import com.correx.core.router.DefaultRouterFacade
import com.correx.core.router.RouterContextBuilder
import com.correx.core.router.RouterFacade
@@ -24,18 +24,13 @@ import com.correx.core.router.model.RouterResponse
import com.correx.core.router.model.RouterState
import com.correx.core.router.model.TurnRole
import com.correx.core.router.model.WorkflowStatus
import com.correx.core.context.model.ContextPack
import com.correx.core.events.types.ContextPackId
import com.correx.core.events.types.ContextEntryId
import com.correx.core.context.model.ContextEntry
import com.correx.core.context.model.ContextLayer
import kotlinx.coroutines.runBlocking
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertFalse
import org.junit.jupiter.api.Assertions.assertNotNull
import org.junit.jupiter.api.Assertions.assertNull
import org.junit.jupiter.api.Assertions.assertTrue
import org.junit.jupiter.api.Test
import kotlinx.coroutines.runBlocking
class RouterFacadeTest {
@@ -44,7 +39,7 @@ class RouterFacadeTest {
// --------------------------------------------------------------------------
@Test
fun `CHAT mode returns inference response content`() = runBlocking {
fun `CHAT mode returns inference response content`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.CHAT)
val response = facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello, world!")
@@ -52,7 +47,7 @@ class RouterFacadeTest {
}
@Test
fun `CHAT mode sets steeringEmitted to false`() = runBlocking {
fun `CHAT mode sets steeringEmitted to false`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.CHAT)
val response = facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!")
@@ -60,7 +55,7 @@ class RouterFacadeTest {
}
@Test
fun `CHAT mode does not append to EventStore`() = runBlocking {
fun `CHAT mode does not append to EventStore`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.CHAT)
facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!")
@@ -72,7 +67,7 @@ class RouterFacadeTest {
// --------------------------------------------------------------------------
@Test
fun `STEERING mode sets steeringEmitted to true`() = runBlocking {
fun `STEERING mode sets steeringEmitted to true`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.STEERING)
val response = facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!")
@@ -80,7 +75,7 @@ class RouterFacadeTest {
}
@Test
fun `STEERING mode appends SteeringNoteAddedEvent with correct session id and user input`() = runBlocking {
fun `STEERING mode appends SteeringNoteAddedEvent with correct session id and user input`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.STEERING)
facade.onUserInput(sessionId = SessionId("session-xyz"), input = "steer this way")
@@ -93,13 +88,17 @@ class RouterFacadeTest {
}
@Test
fun `STEERING mode appends SteeringNoteAddedEvent with stageId from state`() = runBlocking {
fun `STEERING mode appends SteeringNoteAddedEvent with stageId from state`(): Unit = runBlocking {
val mockStore = mockEventStore()
val stageId = StageId("stage-A")
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
override suspend fun getRouterState(sessionId: SessionId): RouterState =
RouterState(sessionId = sessionId, workflowStatus = WorkflowStatus.RUNNING, currentStageId = stageId)
RouterState(
sessionId = sessionId,
workflowStatus = WorkflowStatus.RUNNING,
currentStageId = stageId,
)
},
routerContextBuilder = object : RouterContextBuilder {
override fun build(state: RouterState, budget: TokenBudget): ContextPack = emptyContextPack()
@@ -114,7 +113,7 @@ class RouterFacadeTest {
}
@Test
fun `STEERING mode appends SteeringNoteAddedEvent with null stageId when state has none`() = runBlocking {
fun `STEERING mode appends SteeringNoteAddedEvent with null stageId when state has none`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
@@ -138,12 +137,16 @@ class RouterFacadeTest {
// --------------------------------------------------------------------------
@Test
fun `conversation history grows per call user and router turns appended`() = runBlocking {
fun `conversation history grows per call - user and router turns appended`(): Unit = runBlocking {
val capturedStates = mutableListOf<RouterState>()
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
override suspend fun getRouterState(sessionId: SessionId): RouterState =
RouterState(sessionId = sessionId, workflowStatus = WorkflowStatus.RUNNING, currentStageId = StageId("s1"))
RouterState(
sessionId = sessionId,
workflowStatus = WorkflowStatus.RUNNING,
currentStageId = StageId("s1"),
)
},
routerContextBuilder = object : RouterContextBuilder {
override fun build(state: RouterState, budget: TokenBudget): ContextPack {
@@ -171,7 +174,7 @@ class RouterFacadeTest {
}
@Test
fun `conversation history is session-scoped different sessions do not share history`() = runBlocking {
fun `conversation history is session-scoped - different sessions do not share history`(): Unit = runBlocking {
val capturedStates = mutableListOf<RouterState>()
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
@@ -204,7 +207,7 @@ class RouterFacadeTest {
// --------------------------------------------------------------------------
@Test
fun `state is passed through to context builder`() = runBlocking {
fun `state is passed through to context builder`(): Unit = runBlocking {
val capturedState = mutableListOf<RouterState>()
val mockContextBuilder = object : RouterContextBuilder {
override fun build(state: RouterState, budget: TokenBudget): ContextPack {
@@ -215,7 +218,11 @@ class RouterFacadeTest {
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
override suspend fun getRouterState(sessionId: SessionId): RouterState =
RouterState(sessionId = sessionId, workflowStatus = WorkflowStatus.RUNNING, currentStageId = StageId("s1"))
RouterState(
sessionId = sessionId,
workflowStatus = WorkflowStatus.RUNNING,
currentStageId = StageId("s1"),
)
},
routerContextBuilder = mockContextBuilder,
inferenceRouter = mockInferenceRouter("inference response"),
@@ -229,7 +236,7 @@ class RouterFacadeTest {
}
@Test
fun `budget is passed through to context builder`() = runBlocking {
fun `budget is passed through to context builder`(): Unit = runBlocking {
val capturedBudget = mutableListOf<TokenBudget>()
val mockContextBuilder = object : RouterContextBuilder {
override fun build(state: RouterState, budget: TokenBudget): ContextPack {
@@ -252,10 +259,13 @@ class RouterFacadeTest {
}
@Test
fun `stage ID uses state currentStageId`() = runBlocking {
fun `stage ID uses state currentStageId`(): Unit = runBlocking {
val capturedStageId = mutableListOf<StageId>()
val mockInferenceRouter = object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider {
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider {
capturedStageId.add(stageId)
return mockProvider("response")
}
@@ -281,10 +291,13 @@ class RouterFacadeTest {
}
@Test
fun `stage ID falls back to StageId("none") when state has no currentStageId`() = runBlocking {
fun `stage ID falls back to StageId none when state has no currentStageId`(): Unit = runBlocking {
val capturedStageId = mutableListOf<StageId>()
val mockInferenceRouter = object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider {
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider {
capturedStageId.add(stageId)
return mockProvider("response")
}
@@ -309,10 +322,13 @@ class RouterFacadeTest {
}
@Test
fun `new InferenceRequestId per call`() = runBlocking {
fun `new InferenceRequestId per call`(): Unit = runBlocking {
val capturedRequestIds = mutableListOf<InferenceRequestId>()
val mockInferenceRouter = object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider {
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider {
return mockProviderWithCapture("response", capturedRequestIds)
}
}
@@ -334,10 +350,13 @@ class RouterFacadeTest {
}
@Test
fun `GenerationConfig defaults temperature 0 7 topP 0 9 maxTokens 512`() = runBlocking {
fun `GenerationConfig defaults temperature 0 7 topP 0 9 maxTokens 512`(): Unit = runBlocking {
val capturedRequests = mutableListOf<InferenceRequest>()
val mockInferenceRouter = object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider {
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider {
return mockProviderWithRequestCapture("response", capturedRequests)
}
}
@@ -360,7 +379,7 @@ class RouterFacadeTest {
}
@Test
fun `context pack is passed to inference provider`() = runBlocking {
fun `context pack is passed to inference provider`(): Unit = runBlocking {
val capturedContextPacks = mutableListOf<ContextPack>()
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
@@ -381,7 +400,10 @@ class RouterFacadeTest {
}
},
inferenceRouter = object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider {
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider {
return object : InferenceProvider {
override val id = com.correx.core.events.types.ProviderId("mock")
override val name = "Mock"
@@ -397,8 +419,10 @@ class RouterFacadeTest {
latencyMs = 0,
)
}
override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth =
com.correx.core.inference.ProviderHealth.Healthy
override fun capabilities(): Set<com.correx.core.inference.CapabilityScore> = emptySet()
}
}
@@ -410,7 +434,7 @@ class RouterFacadeTest {
}
@Test
fun `responseFormat defaults to Text`() = runBlocking {
fun `responseFormat defaults to Text`(): Unit = runBlocking {
val capturedRequests = mutableListOf<InferenceRequest>()
val facade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
@@ -420,7 +444,10 @@ class RouterFacadeTest {
override fun build(state: RouterState, budget: TokenBudget): ContextPack = emptyContextPack()
},
inferenceRouter = object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider {
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider {
return mockProviderWithRequestCapture("response", capturedRequests)
}
},
@@ -433,7 +460,7 @@ class RouterFacadeTest {
}
@Test
fun `onUserInput returns RouterResponse with content and mode flag`() = runBlocking {
fun `onUserInput returns RouterResponse with content and mode flag`(): Unit = runBlocking {
val mockStore = mockEventStore()
val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.STEERING)
val response = facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!")
@@ -455,7 +482,11 @@ class RouterFacadeTest {
): RouterFacade = DefaultRouterFacade(
routerRepository = object : RouterRepository {
override suspend fun getRouterState(sessionId: SessionId): RouterState =
RouterState(sessionId = sessionId, workflowStatus = WorkflowStatus.RUNNING, currentStageId = StageId("s1"))
RouterState(
sessionId = sessionId,
workflowStatus = WorkflowStatus.RUNNING,
currentStageId = StageId("s1"),
)
},
routerContextBuilder = object : RouterContextBuilder {
override fun build(state: RouterState, budget: TokenBudget): ContextPack = emptyContextPack()
@@ -472,7 +503,10 @@ class RouterFacadeTest {
private fun mockInferenceRouter(responseText: String): InferenceRouter =
object : InferenceRouter {
override suspend fun route(stageId: StageId, requiredCapabilities: Set<ModelCapability>): InferenceProvider =
override suspend fun route(
stageId: StageId,
requiredCapabilities: Set<ModelCapability>,
): InferenceProvider =
mockProvider(responseText)
}
@@ -488,8 +522,10 @@ class RouterFacadeTest {
tokensUsed = TokenUsage(promptTokens = 10, completionTokens = 5),
latencyMs = 0,
)
override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth =
com.correx.core.inference.ProviderHealth.Healthy
override fun capabilities(): Set<com.correx.core.inference.CapabilityScore> =
setOf(com.correx.core.inference.CapabilityScore(ModelCapability.General, 1.0))
}
@@ -512,8 +548,10 @@ class RouterFacadeTest {
latencyMs = 0,
)
}
override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth =
com.correx.core.inference.ProviderHealth.Healthy
override fun capabilities(): Set<com.correx.core.inference.CapabilityScore> = emptySet()
}
@@ -535,8 +573,10 @@ class RouterFacadeTest {
latencyMs = 0,
)
}
override suspend fun healthCheck(): com.correx.core.inference.ProviderHealth =
com.correx.core.inference.ProviderHealth.Healthy
override fun capabilities(): Set<com.correx.core.inference.CapabilityScore> = emptySet()
}
@@ -568,16 +608,19 @@ class RouterFacadeTest {
override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> =
events.map { append(it) }
override fun read(sessionId: com.correx.core.events.types.SessionId): List<StoredEvent> =
override fun read(sessionId: SessionId): List<StoredEvent> =
storedEvents.values.filter { it.metadata.sessionId == sessionId }.toList()
override fun readFrom(sessionId: com.correx.core.events.types.SessionId, fromSequence: Long): List<StoredEvent> =
override fun readFrom(
sessionId: SessionId,
fromSequence: Long,
): List<StoredEvent> =
read(sessionId).filter { it.sequence >= fromSequence }
override fun lastSequence(sessionId: com.correx.core.events.types.SessionId): Long? =
override fun lastSequence(sessionId: SessionId): Long? =
read(sessionId).maxOfOrNull { it.sequence }
override fun subscribe(sessionId: com.correx.core.events.types.SessionId): kotlinx.coroutines.flow.Flow<StoredEvent> =
override fun subscribe(sessionId: SessionId): kotlinx.coroutines.flow.Flow<StoredEvent> =
throw UnsupportedOperationException("subscribe not implemented for mock")
override fun allEvents(): Sequence<StoredEvent> =
@@ -7,13 +7,13 @@ import com.correx.core.events.events.ToolInvokedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.router.DefaultRouterReducer
import com.correx.core.router.RouterProjector
import com.correx.core.router.model.RouterState
import com.correx.core.router.model.StageOutcomeKind
import com.correx.core.router.model.WorkflowStatus
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.testing.fixtures.EventFixtures.stored
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNull
@@ -26,6 +26,7 @@ class RouterProjectorTest {
private val projector = RouterProjector(reducer)
private val sessionId = SessionId("s1")
private val stageId = StageId("stage-1")
private val workflowId = "workflow-test"
@Test
fun `initial() returns RouterState with IDLE status and empty collections`() {
@@ -49,7 +50,7 @@ class RouterProjectorTest {
val state = projector.initial()
val updated = projector.apply(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
assertEquals(sessionId, updated.sessionId)
assertEquals(WorkflowStatus.RUNNING, updated.workflowStatus)
@@ -60,11 +61,11 @@ class RouterProjectorTest {
fun `apply(WorkflowCompletedEvent) delegates to reducer and sets COMPLETED`() {
val started = projector.apply(
projector.initial(),
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val updated = projector.apply(
started,
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3)),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3, workflowId)),
)
assertEquals(WorkflowStatus.COMPLETED, updated.workflowStatus)
assertNull(updated.currentStageId)
@@ -75,7 +76,7 @@ class RouterProjectorTest {
fun `apply(WorkflowFailedEvent) delegates to reducer and sets FAILED`() {
val started = projector.apply(
projector.initial(),
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val updated = projector.apply(
started,
@@ -89,7 +90,7 @@ class RouterProjectorTest {
fun `apply(OrchestrationPausedEvent) delegates to reducer and sets PAUSED`() {
val started = projector.apply(
projector.initial(),
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val updated = projector.apply(
started,
@@ -103,7 +104,7 @@ class RouterProjectorTest {
fun `apply(OrchestrationResumedEvent) delegates to reducer and sets RUNNING`() {
val started = projector.apply(
projector.initial(),
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val paused = projector.apply(
started,
@@ -133,7 +134,7 @@ class RouterProjectorTest {
fun `apply(StageFailedEvent) delegates to reducer and appends FAILURE entry`() {
val state = projector.apply(
projector.initial(),
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val updated = projector.apply(
state,
@@ -169,13 +170,13 @@ class RouterProjectorTest {
fun `apply produces same result as reducer reduce for every handled event`() {
val state = projector.initial()
val events = listOf(
stored(payload = WorkflowStartedEvent(sessionId, StageId("a"))),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, StageId("a"))),
stored(payload = StageCompletedEvent(sessionId, StageId("a"), StageId("t1"))),
stored(payload = OrchestrationPausedEvent(sessionId, StageId("a"), "hold")),
stored(payload = OrchestrationResumedEvent(sessionId, StageId("a"))),
stored(payload = StageCompletedEvent(sessionId, StageId("b"), StageId("t2"))),
stored(payload = StageFailedEvent(sessionId, StageId("b"), StageId("t3"), "error")),
stored(payload = WorkflowCompletedEvent(sessionId, StageId("c"), 3)),
stored(payload = WorkflowCompletedEvent(sessionId, StageId("c"), 3, workflowId)),
)
var projected = state
@@ -193,7 +194,7 @@ class RouterProjectorTest {
state = projector.apply(
state,
stored(payload = WorkflowStartedEvent(sessionId, StageId("stage-A"))),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, StageId("stage-A"))),
)
assertEquals(WorkflowStatus.RUNNING, state.workflowStatus)
@@ -205,7 +206,7 @@ class RouterProjectorTest {
state = projector.apply(
state,
stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1)),
stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1, workflowId)),
)
assertEquals(WorkflowStatus.COMPLETED, state.workflowStatus)
assertNull(state.currentStageId)
@@ -1,17 +1,17 @@
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.ToolInvokedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.router.DefaultRouterReducer
import com.correx.core.router.model.StageOutcomeKind
import com.correx.core.router.model.WorkflowStatus
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.testing.fixtures.EventFixtures.stored
import kotlinx.datetime.Instant
import org.junit.jupiter.api.Assertions.assertEquals
@@ -25,6 +25,7 @@ class RouterReducerTest {
private val reducer = DefaultRouterReducer()
private val sessionId = SessionId("s1")
private val stageId = StageId("stage-1")
private val workflowId = "workflow-test"
@Test
fun `initial state has IDLE status and empty collections`() {
@@ -41,7 +42,7 @@ class RouterReducerTest {
val state = reducer.initial
val updated = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
assertEquals(sessionId, updated.sessionId)
assertEquals(WorkflowStatus.RUNNING, updated.workflowStatus)
@@ -53,11 +54,11 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val completed = reducer.reduce(
started,
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3)),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3, workflowId)),
)
assertEquals(WorkflowStatus.COMPLETED, completed.workflowStatus)
assertNull(completed.currentStageId)
@@ -69,7 +70,7 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val failed = reducer.reduce(
started,
@@ -84,7 +85,7 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val paused = reducer.reduce(
started,
@@ -99,7 +100,7 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val paused = reducer.reduce(
started,
@@ -130,7 +131,7 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val updated = reducer.reduce(
started,
@@ -146,7 +147,7 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val storedEvent = stored(payload = StageFailedEvent(sessionId, stageId, StageId("t1"), "timeout"))
val updated = reducer.reduce(started, storedEvent)
@@ -189,7 +190,7 @@ class RouterReducerTest {
val initial = reducer.initial
val started = reducer.reduce(
initial,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val withNote = reducer.reduce(
started,
@@ -223,7 +224,7 @@ class RouterReducerTest {
val s0 = reducer.initial
assertEquals(WorkflowStatus.IDLE, s0.workflowStatus)
val s1 = reducer.reduce(s0, stored(payload = WorkflowStartedEvent(sessionId, StageId("stage-A"))))
val s1 = reducer.reduce(s0, stored(payload = WorkflowStartedEvent(sessionId, workflowId, StageId("stage-A"))))
assertEquals(WorkflowStatus.RUNNING, s1.workflowStatus)
assertEquals(StageId("stage-A"), s1.currentStageId)
@@ -231,11 +232,15 @@ class RouterReducerTest {
assertEquals(1, s2.l2Memory.size)
assertEquals(StageOutcomeKind.SUCCESS, s2.l2Memory[0].outcome)
val s3 = reducer.reduce(s2, stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1)))
val s3 =
reducer.reduce(s2, stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1, workflowId)))
assertEquals(WorkflowStatus.COMPLETED, s3.workflowStatus)
assertNull(s3.currentStageId)
val s4 = reducer.reduce(s3, stored(payload = OrchestrationPausedEvent(sessionId, StageId("stage-A"), "manual pause")))
val s4 = reducer.reduce(
s3,
stored(payload = OrchestrationPausedEvent(sessionId, StageId("stage-A"), "manual pause")),
)
assertEquals(WorkflowStatus.PAUSED, s4.workflowStatus)
val s5 = reducer.reduce(s4, stored(payload = OrchestrationResumedEvent(sessionId, StageId("stage-A"))))
@@ -11,7 +11,6 @@ import com.correx.core.events.types.InferenceRequestId
import com.correx.core.events.types.ProviderId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.utils.TypeId
import com.correx.core.inference.FinishReason
import com.correx.core.inference.GenerationConfig
import com.correx.core.inference.InferenceProvider
@@ -20,6 +19,7 @@ import com.correx.core.inference.InferenceResponse
import com.correx.core.inference.InferenceRouter
import com.correx.core.inference.ModelCapability
import com.correx.core.inference.TokenUsage
import com.correx.core.utils.TypeId
import com.correx.testing.fixtures.inference.MockInferenceProvider
import kotlinx.datetime.Clock
import java.util.*
@@ -20,6 +20,7 @@ object WorkflowFixtures {
)
return WorkflowGraph(
id = "workflow-test",
stages = mapOf(a to StageConfig(), b to StageConfig()),
transitions = setOf(t1),
start = a
@@ -32,12 +32,13 @@ object TransitionFixtures {
val c = StageId("C")
return WorkflowGraph(
id = "workflow-test",
stages = mapOf(a to StageConfig(), b to StageConfig(), c to StageConfig()),
transitions = setOf(
TransitionEdge(id = TransitionId("t1"), from = a, to = b, condition = { true }),
TransitionEdge(id = TransitionId("t2"), from = b, to = c, condition = { true }),
),
start = a
start = a,
)
}
@@ -5,13 +5,18 @@ import com.correx.core.approvals.domain.DefaultApprovalEngine
import com.correx.core.approvals.model.ApprovalContext
import com.correx.core.approvals.model.ApprovalDecision
import com.correx.core.approvals.model.ApprovalScopeIdentity
import com.correx.core.artifacts.DefaultArtifactReducer
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.InferenceCompletedEvent
import com.correx.core.events.events.InferenceStartedEvent
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.execution.RetryPolicy
import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.inference.InferenceRepository
@@ -32,23 +37,19 @@ import com.correx.core.sessions.DefaultSessionRepository
import com.correx.core.sessions.projections.replay.DefaultEventReplayer
import com.correx.core.sessions.projections.replay.EventReplayer
import com.correx.core.transitions.graph.StageConfig
import com.correx.core.transitions.graph.TransitionEdge
import com.correx.core.transitions.graph.WorkflowGraph
import com.correx.core.utils.TypeId
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.contracts.fixtures.artifactstore.NoopArtifactStore
import com.correx.testing.fixtures.InferenceFixtures
import com.correx.testing.fixtures.context.ContextFixtures
import com.correx.testing.fixtures.cyclePolicyMissingValidator
import com.correx.testing.fixtures.inference.MockInferenceProvider
import com.correx.testing.fixtures.transitions.TransitionFixtures
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.InferenceCompletedEvent
import com.correx.core.events.events.InferenceStartedEvent
import com.correx.core.events.types.ArtifactId
import com.correx.core.utils.TypeId
import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore
import com.correx.testing.fixtures.transitions.TransitionFixtures.threeStageGraph
import com.correx.testing.kernel.MockSessionEventReplayer
import kotlinx.coroutines.launch
@@ -209,9 +210,10 @@ class SessionOrchestratorIntegrationTest {
// Single-stage graph: enterStage(A) triggers exactly one approval; after resume,
// the always-true transition moves to the terminal "done" and the workflow completes.
val graph = WorkflowGraph(
id = "workflow-test",
stages = mapOf(StageId("A") to StageConfig()),
transitions = setOf(
com.correx.core.transitions.graph.TransitionEdge(
TransitionEdge(
id = com.correx.core.events.types.TransitionId("t1"),
from = StageId("A"),
to = StageId("done"),
@@ -268,6 +270,7 @@ class SessionOrchestratorIntegrationTest {
@Test
fun `workflow with empty graph fails immediately`(): Unit = runBlocking {
val emptyGraph = WorkflowGraph(
id = "workflow-test",
stages = mapOf(StageId("start") to StageConfig()),
transitions = emptySet(),
start = StageId("start"),
@@ -310,8 +313,8 @@ class SessionOrchestratorIntegrationTest {
assertTrue(recordingStore.puts.size >= 2, "Expected at least 2 puts, got ${recordingStore.puts.size}")
val events = eventStore.read(sessionId)
val started = events.mapNotNull { it.payload as? InferenceStartedEvent }.first()
val completed = events.mapNotNull { it.payload as? InferenceCompletedEvent }.first()
val started = events.firstNotNullOf { it.payload as? InferenceStartedEvent }
val completed = events.firstNotNullOf { it.payload as? InferenceCompletedEvent }
assertEquals(recordingStore.idForIndex(0), started.promptArtifactId)
assertEquals(recordingStore.idForIndex(1), completed.responseArtifactId)
@@ -325,17 +328,18 @@ class SessionOrchestratorIntegrationTest {
val z = StageId("Z")
// ghost is missing from stages map but has an outgoing transition so it is non-terminal
val brokenGraph = WorkflowGraph(
id = "workflow-test",
stages = mapOf(a to StageConfig(), b to StageConfig()),
transitions = setOf(
com.correx.core.transitions.graph.TransitionEdge(
TransitionEdge(
id = com.correx.core.events.types.TransitionId("t1"), from = a, to = b,
condition = { true },
),
com.correx.core.transitions.graph.TransitionEdge(
TransitionEdge(
id = com.correx.core.events.types.TransitionId("t2"), from = b, to = ghost,
condition = { true },
),
com.correx.core.transitions.graph.TransitionEdge(
TransitionEdge(
id = com.correx.core.events.types.TransitionId("t3"), from = ghost, to = z,
condition = { true },
),
@@ -391,5 +395,7 @@ private class RecordingArtifactStore : ArtifactStore {
override suspend fun get(id: ArtifactId): ByteArray? = byId[id]
override suspend fun flushBefore(commit: suspend () -> Unit) { commit() }
override suspend fun flushBefore(commit: suspend () -> Unit) {
commit()
}
}
@@ -59,6 +59,7 @@ class ValidationPipelineIntegrationTest {
val a = StageId("A")
val ghost = StageId("GHOST")
val graphWithDanglingTransition = WorkflowGraph(
id = "workflow-test",
stages = mapOf(a to StageConfig()),
transitions = setOf(TransitionEdge(TransitionId("t1"), from = a, to = ghost) { true }),
start = a,
@@ -21,6 +21,7 @@ class OrchestrationProjectorTest {
private val projector = OrchestrationProjector(reducer)
private val sessionId = SessionId("s1")
private val stageId = StageId("stage-1")
private val workflowId = "workflow-test"
@Test
fun `should initialize orchestration with IDLE status`() {
@@ -32,7 +33,7 @@ class OrchestrationProjectorTest {
fun `should set RUNNING on WorkflowStartedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
assertEquals(OrchestrationStatus.RUNNING, started.status)
}
@@ -40,7 +41,7 @@ class OrchestrationProjectorTest {
fun `should set RUNNING after resume on pause`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
assertEquals(OrchestrationStatus.RUNNING, started.status)
val paused =
projector.apply(started, stored(payload = OrchestrationPausedEvent(sessionId, stageId, "Tool approval")))
@@ -56,7 +57,7 @@ class OrchestrationProjectorTest {
fun `should set FAILED on WorkflowFailedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
val failed =
projector.apply(
started,
@@ -69,7 +70,7 @@ class OrchestrationProjectorTest {
fun `same events produce same result`() {
val initialState = projector.initial()
val events = listOf(
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
stored(payload = WorkflowFailedEvent(sessionId, stageId, "Something went wrong", false)),
stored(
payload = RetryAttemptedEvent(
@@ -80,7 +81,7 @@ class OrchestrationProjectorTest {
failureReason = "Something went wrong",
),
),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)),
)
val result1 = events.fold(initialState) { state, event ->
@@ -96,11 +97,11 @@ class OrchestrationProjectorTest {
@Test
fun `full pause-resume lifecycle`() {
val s0 = projector.initial()
val s1 = projector.apply(s0, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
val s1 = projector.apply(s0, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
val s2 =
projector.apply(s1, stored(payload = OrchestrationPausedEvent(sessionId, stageId, "approval required")))
val s3 = projector.apply(s2, stored(payload = OrchestrationResumedEvent(sessionId, stageId)))
val s4 = projector.apply(s3, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)))
val s4 = projector.apply(s3, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)))
assertEquals(OrchestrationStatus.RUNNING, s1.status)
assertEquals(OrchestrationStatus.PAUSED, s2.status)
@@ -22,12 +22,13 @@ class OrchestrationReducerTest {
private val sessionId = SessionId("s1")
private val stageId = StageId("stage-1")
private val state = OrchestrationState(stageId)
private val workflowId = "workflow-test"
@Test
fun `WorkflowStartedEvent sets status to RUNNING`() {
val state = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
assertEquals(OrchestrationStatus.RUNNING, state.status)
assertEquals(stageId, state.currentStageId)
@@ -37,7 +38,7 @@ class OrchestrationReducerTest {
fun `WorkflowFailedEvent sets status to FAILED and failure reason`() {
val started = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val failed = reducer.reduce(
@@ -52,12 +53,12 @@ class OrchestrationReducerTest {
fun `WorkflowCompletedEvent sets status to COMPLETED`() {
val started = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val completed = reducer.reduce(
started,
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)),
)
assertTrue(completed.failureReason.isNullOrBlank())
assertEquals(OrchestrationStatus.COMPLETED, completed.status)
@@ -68,7 +69,7 @@ class OrchestrationReducerTest {
fun `OrchestrationPausedEvent sets status to PAUSED with reason and pending approval`() {
val started = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val paused = reducer.reduce(
@@ -84,7 +85,7 @@ class OrchestrationReducerTest {
fun `OrchestrationResumedEvent sets status to RUNNING with null reason and no pending approval`() {
val started = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val paused = reducer.reduce(
@@ -105,7 +106,7 @@ class OrchestrationReducerTest {
fun `RetryAttemptedEvent sets status to RUNNING with null reason and no pending approval`() {
val started = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val failed = reducer.reduce(
@@ -137,7 +138,7 @@ class OrchestrationReducerTest {
fun `unrelated event does nothing`() {
val started = reducer.reduce(
state,
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
)
val unrelated = reducer.reduce(started, stored())
@@ -1,17 +1,17 @@
import com.correx.core.events.events.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.SteeringNoteAddedEvent
import com.correx.core.events.events.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.core.router.DefaultRouterReducer
import com.correx.core.router.RouterProjector
import com.correx.core.router.model.StageOutcomeKind
import com.correx.core.router.model.WorkflowStatus
import com.correx.core.events.types.SessionId
import com.correx.core.events.types.StageId
import com.correx.testing.fixtures.EventFixtures.stored
import org.junit.jupiter.api.Assertions.assertEquals
import org.junit.jupiter.api.Assertions.assertNull
@@ -23,6 +23,7 @@ class RouterProjectorTest {
private val projector = RouterProjector(reducer)
private val sessionId = SessionId("s1")
private val stageId = StageId("stage-1")
private val workflowId = "workflow-test"
@Test
fun `initial state has IDLE status and empty collections`() {
@@ -37,7 +38,7 @@ class RouterProjectorTest {
fun `should set RUNNING on WorkflowStartedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
assertEquals(WorkflowStatus.RUNNING, started.workflowStatus)
assertEquals(stageId, started.currentStageId)
}
@@ -46,9 +47,9 @@ class RouterProjectorTest {
fun `should set COMPLETED on WorkflowCompletedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
val completed =
projector.apply(started, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)))
projector.apply(started, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)))
assertEquals(WorkflowStatus.COMPLETED, completed.workflowStatus)
assertNull(completed.currentStageId)
}
@@ -57,7 +58,7 @@ class RouterProjectorTest {
fun `should set FAILED on WorkflowFailedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
val failed =
projector.apply(
started,
@@ -71,7 +72,7 @@ class RouterProjectorTest {
fun `should set PAUSED on OrchestrationPausedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
assertEquals(WorkflowStatus.RUNNING, started.workflowStatus)
val paused =
projector.apply(started, stored(payload = OrchestrationPausedEvent(sessionId, stageId, "Tool approval")))
@@ -82,7 +83,7 @@ class RouterProjectorTest {
fun `should set RUNNING after resume on pause`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
val paused =
projector.apply(started, stored(payload = OrchestrationPausedEvent(sessionId, stageId, "Tool approval")))
assertEquals(WorkflowStatus.PAUSED, paused.workflowStatus)
@@ -105,7 +106,7 @@ class RouterProjectorTest {
fun `should append l2Memory entry and clear currentStageId on StageFailedEvent`() {
val initialState = projector.initial()
val started =
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
projector.apply(initialState, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
assertEquals(stageId, started.currentStageId)
val failed =
projector.apply(started, stored(payload = StageFailedEvent(sessionId, stageId, StageId("t1"), "timeout")))
@@ -129,11 +130,11 @@ class RouterProjectorTest {
fun `same events produce same result (determinism)`() {
val initialState = projector.initial()
val events = listOf(
stored(payload = WorkflowStartedEvent(sessionId, stageId)),
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
stored(payload = StageCompletedEvent(sessionId, StageId("stage-2"), StageId("t1"))),
stored(payload = SteeringNoteAddedEvent(sessionId, "note")),
stored(payload = StageFailedEvent(sessionId, stageId, StageId("t2"), "timeout")),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 2)),
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 2, workflowId)),
)
val result1 = events.fold(initialState) { state, event ->
@@ -149,11 +150,12 @@ class RouterProjectorTest {
@Test
fun `full lifecycle stage complete then stage fail then resume then complete`() {
val s0 = projector.initial()
val s1 = projector.apply(s0, stored(payload = WorkflowStartedEvent(sessionId, stageId)))
val s2 = projector.apply(s1, stored(payload = StageCompletedEvent(sessionId, StageId("stage-2"), StageId("t1"))))
val s1 = projector.apply(s0, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
val s2 =
projector.apply(s1, stored(payload = StageCompletedEvent(sessionId, StageId("stage-2"), StageId("t1"))))
val s3 = projector.apply(s2, stored(payload = StageFailedEvent(sessionId, stageId, StageId("t2"), "timeout")))
val s4 = projector.apply(s3, stored(payload = WorkflowStartedEvent(sessionId, StageId("stage-3"))))
val s5 = projector.apply(s4, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1)))
val s4 = projector.apply(s3, stored(payload = WorkflowStartedEvent(sessionId, workflowId, StageId("stage-3"))))
val s5 = projector.apply(s4, stored(payload = WorkflowCompletedEvent(sessionId, stageId, 1, workflowId)))
assertEquals(WorkflowStatus.RUNNING, s1.workflowStatus)
assertEquals(StageId("stage-2"), s2.l2Memory[0].stageId)
@@ -169,7 +171,8 @@ class RouterProjectorTest {
@Test
fun `unknown event type returns unchanged state`() {
val initialState = projector.initial()
val updated = projector.apply(initialState, stored(payload = com.correx.core.events.events.ToolInvokedEvent("test")))
val updated =
projector.apply(initialState, stored(payload = com.correx.core.events.events.ToolInvokedEvent("test")))
assertEquals(initialState, updated)
}
}
@@ -19,7 +19,7 @@ class DefaultTransitionResolverTest {
private val evaluator = object : TransitionConditionEvaluator {
override fun evaluate(
condition: TransitionCondition,
context: EvaluationContext
context: EvaluationContext,
): Boolean {
return condition.evaluate(context)
}
@@ -36,20 +36,20 @@ class DefaultTransitionResolverTest {
id = "t1",
from = "stage-a",
to = "stage-b",
condition = alwaysTrue()
)
condition = alwaysTrue(),
),
stages = setOf("stage-a", "stage-b")
),
stages = setOf("stage-a", "stage-b"),
)
val result = resolver.resolve(
graph = graph,
context = context(currentStage = "stage-b")
context = context(currentStage = "stage-b"),
)
assertEquals(
TransitionDecision.NoMatch,
result
result,
)
}
@@ -62,19 +62,19 @@ class DefaultTransitionResolverTest {
id = "t1",
from = "stage-a",
to = "stage-b",
condition = alwaysFalse()
)
)
condition = alwaysFalse(),
),
),
)
val result = resolver.resolve(
graph = graph,
context = context(currentStage = "stage-a")
context = context(currentStage = "stage-a"),
)
assertEquals(
TransitionDecision.Stay,
result
result,
)
}
@@ -87,26 +87,26 @@ class DefaultTransitionResolverTest {
id = "t1",
from = "stage-a",
to = "stage-b",
condition = alwaysTrue()
)
)
condition = alwaysTrue(),
),
),
)
val result = resolver.resolve(
graph = graph,
context = context(currentStage = "stage-a")
context = context(currentStage = "stage-a"),
)
val move = assertInstanceOf<TransitionDecision.Move>(result)
assertEquals(
TransitionId("t1"),
move.transitionId
move.transitionId,
)
assertEquals(
StageId("stage-b"),
move.to
move.to,
)
}
@@ -119,32 +119,32 @@ class DefaultTransitionResolverTest {
id = "t2",
from = "stage-a",
to = "stage-c",
condition = alwaysTrue()
condition = alwaysTrue(),
),
edge(
id = "t1",
from = "stage-a",
to = "stage-b",
condition = alwaysTrue()
)
)
condition = alwaysTrue(),
),
),
)
val result = resolver.resolve(
graph = graph,
context = context(currentStage = "stage-a")
context = context(currentStage = "stage-a"),
)
val move = assertInstanceOf<TransitionDecision.Move>(result)
assertEquals(
TransitionId("t1"),
move.transitionId
move.transitionId,
)
assertEquals(
StageId("stage-b"),
move.to
move.to,
)
}
@@ -157,27 +157,27 @@ class DefaultTransitionResolverTest {
id = "t1",
from = "stage-x",
to = "stage-y",
condition = alwaysTrue()
condition = alwaysTrue(),
),
edge(
id = "t2",
from = "stage-a",
to = "stage-b",
condition = alwaysTrue()
)
)
condition = alwaysTrue(),
),
),
)
val result = resolver.resolve(
graph = graph,
context = context(currentStage = "stage-a")
context = context(currentStage = "stage-a"),
)
val move = assertInstanceOf<TransitionDecision.Move>(result)
assertEquals(
TransitionId("t2"),
move.transitionId
move.transitionId,
)
}
@@ -185,24 +185,25 @@ class DefaultTransitionResolverTest {
id: String,
from: String,
to: String,
condition: TransitionCondition
condition: TransitionCondition,
): TransitionEdge {
return TransitionEdge(
id = TransitionId(id),
from = StageId(from),
to = StageId(to),
condition = condition
condition = condition,
)
}
private fun graph(
start: String,
transitions: Set<TransitionEdge>,
stages: Set<String> = transitions.flatMap { listOf(it.from.value, it.to.value) }.toSet()
stages: Set<String> = transitions.flatMap { listOf(it.from.value, it.to.value) }.toSet(),
) = WorkflowGraph(
id = "workflow-test",
start = StageId(start),
stages = stages.associate { StageId(it) to StageConfig() },
transitions = transitions
transitions = transitions,
)
private fun context(