From 219e2c762e8b4adf6080fab3b30d58a12d7fae1e Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 18 May 2026 12:22:38 +0400 Subject: [PATCH] =?UTF-8?q?feat(cas):=20content-addressed=20artifact=20sto?= =?UTF-8?q?re=20(steps=201=E2=80=938)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New core:artifacts-store interface + infrastructure/artifacts-cas implementation: segment files + SQLite index, Blake3 hashing, group-commit fsync via flushBefore, recovery tail-scan, manual compactor, and oldest-first disk-cap evictor. Inference events now carry promptArtifactId / responseArtifactId; orchestrators put bytes before emitting. SqliteEventStore wraps its txn in artifactStore.flushBefore so segment data is fsynced before the event commit, making the crash window non-corrupting (TailScanner re-indexes orphan tail records on reopen). Compactor and evictor are mutually exclusive via maintenanceMutex. Step 9 (cloud sync) deferred to a later epic. See docs/reviews/2026-05-18-cas-steps-1-8-review.md for the final review. --- apps/server/build.gradle | 1 + .../kotlin/com/correx/apps/server/Main.kt | 4 +- .../apps/server/logging/LoggingEventStore.kt | 6 +- .../apps/server/protocol/ServerMessage.kt | 8 +- .../registry/FileSystemWorkflowRegistry.kt | 6 +- .../apps/server/ws/GlobalStreamHandler.kt | 7 +- .../apps/tui/components/ActiveSession.kt | 5 + .../apps/tui/reducer/SessionsReducer.kt | 28 +- .../com/correx/apps/tui/state/TuiState.kt | 1 + .../DefaultApprovalRepositoryTest.kt | 5 +- core/artifacts-store/build.gradle | 9 + .../core/artifactstore/ArtifactStore.kt | 9 + .../com/correx/core/events/EventDispatcher.kt | 2 +- .../core/events/events/InferenceEvents.kt | 3 + .../correx/core/events/stores/EventStore.kt | 10 +- core/kernel/build.gradle | 2 + .../DefaultSessionOrchestrator.kt | 96 +++++-- .../orchestration/ReplayOrchestrator.kt | 51 +++- .../orchestration/SessionOrchestrator.kt | 98 +++++-- .../2026-05-18-cas-steps-1-8-review.md | 62 ++++ infrastructure/artifacts-cas/build.gradle | 16 ++ .../artifactscas/CasArtifactStore.kt | 108 +++++++ .../artifactscas/compact/Compactor.kt | 146 ++++++++++ .../artifactscas/compact/LivenessScanner.kt | 77 +++++ .../artifactscas/config/CasConfig.kt | 14 + .../artifactscas/evict/Evictor.kt | 61 ++++ .../artifactscas/index/ArtifactIndex.kt | 13 + .../artifactscas/index/SqliteArtifactIndex.kt | 136 +++++++++ .../artifactscas/recovery/TailScanner.kt | 75 +++++ .../artifactscas/segment/SegmentLayout.kt | 44 +++ .../artifactscas/segment/SegmentReader.kt | 32 +++ .../artifactscas/segment/SegmentWriter.kt | 74 +++++ .../artifactscas/CasArtifactStoreTest.kt | 101 +++++++ .../artifactscas/compact/CompactorTest.kt | 200 +++++++++++++ .../artifactscas/evict/EvictorTest.kt | 267 ++++++++++++++++++ .../artifactscas/recovery/TailScannerTest.kt | 81 ++++++ infrastructure/build.gradle | 2 + .../llama/cpp/DefaultModelManager.kt | 3 +- .../llama/cpp/LlamaCppInferenceProvider.kt | 6 + infrastructure/persistence/build.gradle | 1 + .../persistence/InMemoryEventStore.kt | 9 +- .../persistence/SqliteEventStore.kt | 122 +++++--- .../persistence/SqliteEventStoreTest.kt | 3 +- .../persistence/SqliteReplayTest.kt | 3 +- .../infrastructure/InfrastructureModule.kt | 24 +- .../workflow/TomlWorkflowLoader.kt | 6 +- settings.gradle | 2 + testing/contracts/build.gradle | 2 + .../artifactstore/NoopArtifactStore.kt | 11 + .../events/store/EventStoreContractTest.kt | 45 +-- .../projections/ReplayContractTest.kt | 3 +- .../projections/SessionReplayContractTest.kt | 5 +- .../kotlin/SessionReplayDeterminismTest.kt | 3 +- .../testing/fixtures/InferenceFixtures.kt | 9 + testing/integration/build.gradle | 2 + .../SessionOrchestratorIntegrationTest.kt | 70 ++++- .../src/test/kotlin/InferenceProjectorTest.kt | 5 + .../src/test/kotlin/InferenceReducerTest.kt | 4 + .../src/test/kotlin/SessionReplayTest.kt | 2 +- .../kotlin/TransitionReplayIntegrationTest.kt | 7 +- 60 files changed, 2042 insertions(+), 165 deletions(-) create mode 100644 core/artifacts-store/build.gradle create mode 100644 core/artifacts-store/src/main/kotlin/com/correx/core/artifactstore/ArtifactStore.kt create mode 100644 docs/reviews/2026-05-18-cas-steps-1-8-review.md create mode 100644 infrastructure/artifacts-cas/build.gradle create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStore.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/Compactor.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/LivenessScanner.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/config/CasConfig.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/evict/Evictor.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/ArtifactIndex.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/SqliteArtifactIndex.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScanner.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentLayout.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentReader.kt create mode 100644 infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentWriter.kt create mode 100644 infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStoreTest.kt create mode 100644 infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/compact/CompactorTest.kt create mode 100644 infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/evict/EvictorTest.kt create mode 100644 infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScannerTest.kt create mode 100644 testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/artifactstore/NoopArtifactStore.kt diff --git a/apps/server/build.gradle b/apps/server/build.gradle index b5ef6485..bc14e809 100644 --- a/apps/server/build.gradle +++ b/apps/server/build.gradle @@ -19,6 +19,7 @@ dependencies { implementation project(':core:validation') implementation project(':core:risk') implementation project(':core:artifacts') + implementation project(':core:artifacts-store') implementation project(':infrastructure') implementation project(':infrastructure:workflow') implementation project(':infrastructure:persistence') diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index f4fdd183..bb165f83 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -31,7 +31,8 @@ import io.ktor.server.engine.embeddedServer import io.ktor.server.netty.Netty fun main() { - val eventStore = LoggingEventStore(InfrastructureModule.createEventStore()) + val artifactStore = InfrastructureModule.createArtifactStore() + val eventStore = LoggingEventStore(InfrastructureModule.createEventStore(artifactStore)) val llamaProvider = InfrastructureModule.createLlamaCppProvider( modelId = System.getenv("CORREX_MODEL_ID") ?: "default", @@ -79,6 +80,7 @@ fun main() { repositories = repositories, engines = engines, retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = artifactStore, ) val workflowRegistry = FileSystemWorkflowRegistry(InfrastructureModule.createWorkflowLoader()) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt b/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt index 8b46a940..c651dcd8 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/logging/LoggingEventStore.kt @@ -11,7 +11,7 @@ private val log = LoggerFactory.getLogger(LoggingEventStore::class.java) class LoggingEventStore(private val delegate: EventStore) : EventStore { - override fun append(event: NewEvent): StoredEvent = + override suspend fun append(event: NewEvent): StoredEvent = runCatching { delegate.append(event) } .onSuccess { stored -> log.info( @@ -31,7 +31,7 @@ class LoggingEventStore(private val delegate: EventStore) : EventStore { } .getOrThrow() - override fun appendAll(events: List): List = + override suspend fun appendAll(events: List): List = runCatching { delegate.appendAll(events) } .onSuccess { stored -> val breakdown = events.groupingBy { it.payload::class.simpleName }.eachCount() @@ -58,4 +58,6 @@ class LoggingEventStore(private val delegate: EventStore) : EventStore { override fun lastSequence(sessionId: SessionId): Long? = delegate.lastSequence(sessionId) override fun subscribe(sessionId: SessionId): Flow = delegate.subscribe(sessionId) + + override fun allEvents(): Sequence = delegate.allEvents() } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index 13f1ff29..919a8caa 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -34,8 +34,12 @@ sealed class ServerMessage { data class InferenceStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() @Serializable - data class InferenceCompleted(val sessionId: SessionId, val stageId: StageId, val outputSummary: String) : - ServerMessage() + data class InferenceCompleted( + val sessionId: SessionId, + val stageId: StageId, + val outputSummary: String, + val responseText: String = "", + ) : ServerMessage() @Serializable data class InferenceTimedOut(val sessionId: SessionId, val stageId: StageId, val elapsedMs: Long) : diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt b/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt index a47d9cc8..37b3d57a 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/registry/FileSystemWorkflowRegistry.kt @@ -12,7 +12,7 @@ private val log = LoggerFactory.getLogger(FileSystemWorkflowRegistry::class.java class FileSystemWorkflowRegistry( private val loader: WorkflowLoader, private val workflowsDir: Path = Path.of( - System.getProperty("user.home"), ".config", "correx", "workflows" + System.getProperty("user.home"), ".config", "correx", "workflows", ), ) : WorkflowRegistry { @@ -32,8 +32,8 @@ class FileSystemWorkflowRegistry( .getOrNull() } .associateBy { graph -> - graph.id.ifBlank { graph.let { it } .let { file -> file.start.value } } - } + graph.id.ifBlank { graph.start.value } + }.also { log.info("cache init done, cache keys: {}", it.keys) } } private fun logWarning(file: Path, ex: Throwable) { diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt index d5283add..4157a6cd 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -48,7 +48,11 @@ class GlobalStreamHandler(private val module: ServerModule) { runCatching { ProtocolSerializer.decodeClientMessage(text) } .onSuccess { msg -> log.debug("recv: {}", msg::class.simpleName) - handleClientMessage(session, msg) + runCatching { + handleClientMessage(session, msg) + }.onFailure { + log.error("handle failed", it) + } } .onFailure { log.warn("decode error: {}", it.message) @@ -90,6 +94,7 @@ class GlobalStreamHandler(private val module: ServerModule) { is ClientMessage.StartSession -> { val graph = module.workflowRegistry.find(msg.workflowId) + log.info("find returned: {}", graph) if (graph == null) { val error = ServerMessage.ProtocolError("Unknown workflowId: ${msg.workflowId}") session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt index 0a5ca52c..93bda4e8 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/ActiveSession.kt @@ -16,6 +16,11 @@ fun activeSessionWidget(session: SessionSummary?): Paragraph { val lines = buildList { add(Line.from(Span.raw("stage: ${session.currentStage ?: "—"}"))) session.lastOutput?.let { add(Line.from(Span.raw("last output: $it"))) } + session.lastResponseText?.takeIf { it.isNotEmpty() }?.let { text -> + add(Line.from(Span.raw(""))) + add(Line.from(Span.raw("response:"))) + text.lineSequence().forEach { add(Line.from(Span.raw(it))) } + } } Text.from(lines) } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt index 6ac93843..b7de4c8d 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt @@ -100,15 +100,7 @@ object SessionsReducer { ) is ServerMessage.StageCompleted -> touchSession(sessions, msg.sessionId.value, null, clock) is ServerMessage.StageFailed -> touchSession(sessions, msg.sessionId.value, null, clock) - is ServerMessage.InferenceCompleted -> sessions.copy( - sessions = sessions.sessions.map { s -> - if (s.id == msg.sessionId.value) { - s.copy(lastOutput = msg.outputSummary, lastEventAt = clock()) - } else { - s - } - }, - ) + is ServerMessage.InferenceCompleted -> applyInferenceCompleted(sessions, msg, clock) is ServerMessage.ToolCompleted -> sessions.copy( sessions = sessions.sessions.map { s -> if (s.id == msg.sessionId.value) { @@ -121,6 +113,24 @@ object SessionsReducer { else -> sessions } + private fun applyInferenceCompleted( + sessions: SessionsState, + msg: ServerMessage.InferenceCompleted, + clock: () -> Long, + ): SessionsState = sessions.copy( + sessions = sessions.sessions.map { s -> + if (s.id == msg.sessionId.value) { + s.copy( + lastOutput = msg.outputSummary, + lastResponseText = msg.responseText.takeIf { it.isNotEmpty() }, + lastEventAt = clock(), + ) + } else { + s + } + }, + ) + private fun touchSession( sessions: SessionsState, sessionId: String, diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt index 527034a7..bfa6d968 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/state/TuiState.kt @@ -17,6 +17,7 @@ data class SessionSummary( val lastEventAt: Long, val currentStage: String?, val lastOutput: String?, + val lastResponseText: String? = null, ) data class ApprovalInfo( diff --git a/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalRepositoryTest.kt b/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalRepositoryTest.kt index ab833c81..525677e5 100644 --- a/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalRepositoryTest.kt +++ b/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalRepositoryTest.kt @@ -12,6 +12,7 @@ import com.correx.core.events.types.SessionId import com.correx.core.events.types.ValidationReportId import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.infrastructure.persistence.InMemoryEventStore +import kotlinx.coroutines.runBlocking import kotlinx.datetime.Instant import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertTrue @@ -41,7 +42,7 @@ class DefaultApprovalRepositoryTest { } @Test - fun `after ApprovalRequestedEvent state contains the request`() { + fun `after ApprovalRequestedEvent state contains the request`(): Unit = runBlocking { val sessionId = SessionId("session-requested") val requestId = ApprovalRequestId("req-1") @@ -66,7 +67,7 @@ class DefaultApprovalRepositoryTest { } @Test - fun `after ApprovalGrantCreatedEvent then ApprovalGrantExpiredEvent grants list is empty`() { + fun `after ApprovalGrantCreatedEvent then ApprovalGrantExpiredEvent grants list is empty`(): Unit = runBlocking { val sessionId = SessionId("session-grant-expired") val grantId = GrantId("grant-1") diff --git a/core/artifacts-store/build.gradle b/core/artifacts-store/build.gradle new file mode 100644 index 00000000..29c06aaa --- /dev/null +++ b/core/artifacts-store/build.gradle @@ -0,0 +1,9 @@ +plugins { + id 'java-library' + id 'org.jetbrains.kotlin.jvm' + id 'org.jetbrains.kotlin.plugin.serialization' +} + +dependencies { + implementation(project(":core:events")) +} diff --git a/core/artifacts-store/src/main/kotlin/com/correx/core/artifactstore/ArtifactStore.kt b/core/artifacts-store/src/main/kotlin/com/correx/core/artifactstore/ArtifactStore.kt new file mode 100644 index 00000000..a6b429a6 --- /dev/null +++ b/core/artifacts-store/src/main/kotlin/com/correx/core/artifactstore/ArtifactStore.kt @@ -0,0 +1,9 @@ +package com.correx.core.artifactstore + +import com.correx.core.events.types.ArtifactId + +interface ArtifactStore { + suspend fun put(bytes: ByteArray): ArtifactId + suspend fun get(id: ArtifactId): ByteArray? + suspend fun flushBefore(commit: suspend () -> Unit) +} diff --git a/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt b/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt index 8960dc2e..c933d38c 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/EventDispatcher.kt @@ -12,7 +12,7 @@ import kotlinx.datetime.Clock import java.util.* class EventDispatcher(private val eventStore: EventStore) { - fun emit( + suspend fun emit( payload: EventPayload, sessionId: SessionId, causationId: CausationId? = null, diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/InferenceEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/InferenceEvents.kt index 39299cff..3638663e 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/InferenceEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/InferenceEvents.kt @@ -1,5 +1,6 @@ package com.correx.core.events.events +import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.InferenceRequestId import com.correx.core.events.types.ProviderId import com.correx.core.events.types.SessionId @@ -14,6 +15,7 @@ data class InferenceStartedEvent( val sessionId: SessionId, val stageId: StageId, val providerId: ProviderId, + val promptArtifactId: ArtifactId, ) : EventPayload @Serializable @@ -24,6 +26,7 @@ data class InferenceCompletedEvent( val providerId: ProviderId, val tokensUsed: TokenUsage, val latencyMs: Long, + val responseArtifactId: ArtifactId, ) : EventPayload @Serializable diff --git a/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt b/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt index 1dad28e8..0c690f18 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/stores/EventStore.kt @@ -14,12 +14,12 @@ interface EventStore { * - idempotency by eventId * - causal consistency rules (if enforced here, otherwise in kernel) */ - fun append(event: NewEvent): StoredEvent + suspend fun append(event: NewEvent): StoredEvent /** * Append multiple events atomically (same session preferred). */ - fun appendAll(events: List): List + suspend fun appendAll(events: List): List /** * Read events for a session in strict sequence order. @@ -41,4 +41,10 @@ interface EventStore { * Does not replay historical events — use [readFrom] for catch-up before subscribing. */ fun subscribe(sessionId: SessionId): Flow + + /** + * Iterate every event in the store across all sessions, in monotonic insertion order. + * Intended for batch jobs (e.g. CAS liveness scans). Implementations should stream lazily. + */ + fun allEvents(): Sequence } \ No newline at end of file diff --git a/core/kernel/build.gradle b/core/kernel/build.gradle index 47879f3e..75139771 100644 --- a/core/kernel/build.gradle +++ b/core/kernel/build.gradle @@ -14,5 +14,7 @@ dependencies { implementation project(':core:inference') implementation project(':core:tools') implementation project(':core:artifacts') + implementation project(':core:artifacts-store') implementation project(':core:risk') + implementation "org.slf4j:slf4j-api:2.0.16" } \ No newline at end of file diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 562c3993..82cd0cc2 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -1,6 +1,8 @@ package com.correx.core.kernel.orchestration import com.correx.core.approvals.model.ApprovalDecision +import com.correx.core.artifactstore.ArtifactStore +import org.slf4j.LoggerFactory import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.SessionId @@ -18,11 +20,14 @@ import java.util.concurrent.atomic.* "ReturnCount", "NestedBlockDepth", ) +private val log = LoggerFactory.getLogger(DefaultSessionOrchestrator::class.java) + class DefaultSessionOrchestrator( private val repositories: OrchestratorRepositories, engines: OrchestratorEngines, private val retryCoordinator: RetryCoordinator, -) : SessionOrchestrator(repositories, engines), ApprovalGateway { + artifactStore: ArtifactStore, +) : SessionOrchestrator(repositories, engines, artifactStore), ApprovalGateway { override val cancellations: ConcurrentHashMap = ConcurrentHashMap() @@ -33,8 +38,15 @@ class DefaultSessionOrchestrator( ): WorkflowResult { System.err.println("[Orchestrator] session=${sessionId.value} workflow=${graph.id} start=${graph.start.value}") emitWorkflowStarted(sessionId, graph, config) + val base = ExecutionContext(graph, sessionId, 0, graph.start, config, null, null) - return step(base.enrich()) + val enriched = base.enrich() + + // Execute the start stage before entering the step loop + return when (val result = enterStage(enriched, graph.start)) { + is StepResult.Continue -> step(result.ctx) + is StepResult.Terminal -> result.result + } } private tailrec suspend fun step(ctx: EnrichedExecutionContext): WorkflowResult { @@ -73,7 +85,12 @@ class DefaultSessionOrchestrator( is TransitionDecision.Stay -> if (isTerminal(enriched.graph, enriched.currentStageId)) { completeWorkflow(enriched.sessionId, enriched.currentStageId, enriched.stageCount) } else { - step(enriched) + failWorkflow( + sessionId = enriched.sessionId, + stageId = enriched.currentStageId, + reason = "no transition condition matched from stage ${enriched.currentStageId.value}", + retryExhausted = false, + ) } is TransitionDecision.Blocked -> failWorkflow( @@ -110,43 +127,60 @@ class DefaultSessionOrchestrator( val nextStageId = decision.to 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 + 1), + completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount), ) } - System.err.println("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${nextStageId.value}") - return when (val result = executeStage(ctx.sessionId, nextStageId, ctx.graph, ctx.session, ctx.config)) { - is StageExecutionResult.Success -> StepResult.Continue( - ctx.copy( + return when (val result = enterStage(ctx, nextStageId)) { + is StepResult.Continue -> StepResult.Continue( + result.ctx.copy( currentStageId = advanceStage(ctx.sessionId, ctx.currentStageId, decision), - stageCount = ctx.stageCount + 1, ), ) + is StepResult.Terminal -> result + } + } - is StageExecutionResult.Failure -> { - System.err.println( - "[Orchestrator] stage failed session=${ctx.sessionId.value} " + - "stage=${nextStageId.value} reason=${result.reason} retryable=${result.retryable}" - ) - val shouldRetry = retryCoordinator.shouldRetry( - sessionId = ctx.sessionId, - stageId = nextStageId, - currentAttempt = ctx.state.retryCount, - policy = ctx.config.retryPolicy, - failureReason = result.reason, - ) - if (shouldRetry) { - StepResult.Continue(ctx) // retryCount updated via projection on next iteration - } else { - StepResult.Terminal( - failWorkflow( - ctx.sessionId, - ctx.currentStageId, - result.reason, - retryExhausted = result.retryable, - ), + @Suppress("ReturnCount") + private suspend fun enterStage( + ctx: EnrichedExecutionContext, + stageId: StageId, + ): StepResult { + log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}") + while (true) { + if (isCancelled(ctx.sessionId)) { + return StepResult.Terminal(handleCancellation(ctx.sessionId, stageId)) + } + when (val result = executeStage(ctx.sessionId, stageId, ctx.graph, ctx.session, ctx.config)) { + is StageExecutionResult.Success -> + return StepResult.Continue(ctx.copy(stageCount = ctx.stageCount + 1)) + + is StageExecutionResult.Failure -> { + log.debug( + "[Orchestrator] stage failed session=${ctx.sessionId.value} " + + "stage=${stageId.value} reason=${result.reason} retryable=${result.retryable}", ) + val refreshedState = orchestrationRepository.getState(ctx.sessionId) + val shouldRetry = retryCoordinator.shouldRetry( + sessionId = ctx.sessionId, + stageId = stageId, + currentAttempt = refreshedState.retryCount, + policy = ctx.config.retryPolicy, + failureReason = result.reason, + ) + if (!shouldRetry) { + return StepResult.Terminal( + failWorkflow( + ctx.sessionId, + stageId, + result.reason, + retryExhausted = result.retryable, + ), + ) + } + // retry — loop and re-execute } } } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt index 2be0ca2c..7e8588a0 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ReplayOrchestrator.kt @@ -1,6 +1,8 @@ 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 @@ -8,6 +10,7 @@ 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.sessions.Session import com.correx.core.inference.InferenceRequest import com.correx.core.kernel.execution.ReplayStrategy import com.correx.core.kernel.execution.WorkflowResult @@ -23,6 +26,8 @@ import java.util.* import java.util.concurrent.* import java.util.concurrent.atomic.* +private val log = LoggerFactory.getLogger(ReplayOrchestrator::class.java) + private data class ReplayContext( val graph: WorkflowGraph, val sessionId: SessionId, @@ -31,6 +36,11 @@ private data class ReplayContext( val config: OrchestrationConfig, ) +private sealed class ReplayStepResult { + data class Continue(val ctx: ReplayContext) : ReplayStepResult() + data class Terminal(val result: WorkflowResult) : ReplayStepResult() +} + /** * Replay is environment-independent. No live risk assessment or approval evaluation — * both are reconstructed from the event log (ApprovalGrantedEvent / ApprovalDeniedEvent). @@ -41,9 +51,11 @@ class ReplayOrchestrator( engines: OrchestratorEngines, override val cancellations: ConcurrentHashMap, private val strategy: ReplayStrategy, + artifactStore: ArtifactStore, ) : SessionOrchestrator( repositories, engines.copy(approvalEngine = NoOpApprovalEngine(), riskAssessor = NoOpRiskAssessor()), + artifactStore, ) { private val replayProvider = ReplayInferenceProvider(repositories.eventStore) @@ -53,7 +65,15 @@ class ReplayOrchestrator( config: OrchestrationConfig, ): WorkflowResult { emitWorkflowStarted(sessionId, graph, config) - return step(ReplayContext(graph, sessionId, graph.start, 0, config)) + + val session = repositories.sessionRepository.getSession(sessionId) + val ctx = ReplayContext(graph, sessionId, graph.start, 0, config) + + // Execute the start stage before entering the step loop + return when (val result = enterStage(ctx, graph.start, session)) { + is ReplayStepResult.Continue -> step(result.ctx) + is ReplayStepResult.Terminal -> result.result + } } private tailrec suspend fun step(ctx: ReplayContext): WorkflowResult { @@ -66,18 +86,16 @@ class ReplayOrchestrator( val nextStageId = decision.to if (isTerminal(ctx.graph, nextStageId)) { - return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount + 1) + return completeWorkflow(ctx.sessionId, nextStageId, ctx.stageCount) } - when (val result = executeStage(ctx.sessionId, nextStageId, ctx.graph, session, ctx.config)) { - is StageExecutionResult.Success -> step( - ctx.copy( + when (val result = enterStage(ctx, nextStageId, session)) { + is ReplayStepResult.Continue -> step( + result.ctx.copy( currentStageId = advanceStage(ctx.sessionId, ctx.currentStageId, decision), - stageCount = ctx.stageCount + 1, ), ) - is StageExecutionResult.Failure -> - failWorkflow(ctx.sessionId, ctx.currentStageId, result.reason, retryExhausted = false) + is ReplayStepResult.Terminal -> return result.result } } @@ -103,6 +121,23 @@ class ReplayOrchestrator( } } + private suspend fun enterStage( + ctx: ReplayContext, + stageId: StageId, + session: Session, + ): ReplayStepResult { + log.debug("[Orchestrator] executeStage session=${ctx.sessionId.value} stage=${stageId.value}") + return when (val result = executeStage(ctx.sessionId, stageId, ctx.graph, session, ctx.config)) { + is StageExecutionResult.Success -> ReplayStepResult.Continue( + ctx.copy(stageCount = ctx.stageCount + 1), + ) + is StageExecutionResult.Failure -> + ReplayStepResult.Terminal( + failWorkflow(ctx.sessionId, stageId, result.reason, retryExhausted = false), + ) + } + } + override suspend fun cancel(sessionId: SessionId) { cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 39c679c7..5d9254b0 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -14,6 +14,7 @@ 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 @@ -57,6 +58,7 @@ import com.correx.core.transitions.resolution.TransitionResolver import com.correx.core.validation.model.ValidationContext import com.correx.core.validation.pipeline.ValidationOutcome import com.correx.core.validation.pipeline.ValidationPipeline +import org.slf4j.LoggerFactory import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.TimeoutCancellationException import kotlinx.coroutines.withTimeout @@ -78,7 +80,9 @@ import kotlin.coroutines.cancellation.CancellationException abstract class SessionOrchestrator( repositories: OrchestratorRepositories, engines: OrchestratorEngines, + private val artifactStore: ArtifactStore, ) { + private val log = LoggerFactory.getLogger(this::class.java) private val eventStore: EventStore = repositories.eventStore private val transitionResolver: TransitionResolver = engines.transitionResolver private val contextPackBuilder: ContextPackBuilder = engines.contextPackBuilder @@ -87,6 +91,7 @@ abstract class SessionOrchestrator( private val riskAssessor: RiskAssessor = engines.riskAssessor private val promptResolver: PromptResolver = engines.promptResolver private val inferenceRepository: InferenceRepository = repositories.inferenceRepository + private val artifactRepository = repositories.artifactRepository internal val orchestrationRepository: OrchestrationRepository = repositories.orchestrationRepository internal abstract val cancellations: ConcurrentHashMap internal val pendingApprovals: ConcurrentHashMap> = @@ -112,12 +117,13 @@ abstract class SessionOrchestrator( val stageConfig = requireNotNull(graph.stages[stageId]) { "Stage '${stageId.value}' not declared in workflow graph" } - val promptEntries = stageConfig.metadata["prompt"] + val systemPrompt = stageConfig.metadata["systemPrompt"] ?.let { path -> runCatching { promptResolver.resolve(path) } .onFailure { - System.err.println( - "[SessionOrchestrator] stage=${stageId.value}: failed to load prompt '$path': ${it.message}" + log.error( + "[SessionOrchestrator] stage=${stageId.value}: " + + "failed to load prompt '$path': ${it.message}", ) } .getOrNull() @@ -135,12 +141,36 @@ abstract class SessionOrchestrator( ), ) } ?: emptyList() + val promptEntries = stageConfig.metadata["prompt"] + ?.let { path -> + runCatching { promptResolver.resolve(path) } + .onFailure { + log.error( + "[SessionOrchestrator] stage=${stageId.value}: " + + "failed to load prompt '$path': ${it.message}", + ) + } + .getOrNull() + } + ?.takeIf { it.isNotBlank() } + ?.let { text -> + listOf( + ContextEntry( + id = ContextEntryId(UUID.randomUUID().toString()), + layer = ContextLayer.L1, + content = text, + sourceType = "agentPrompt", + sourceId = stageId.value, + tokenEstimate = text.length / 4, + ), + ) + } ?: emptyList() val contextPack = contextPackBuilder.build( id = ContextPackId(UUID.randomUUID().toString()), sessionId = sessionId, stageId = stageId, - entries = promptEntries, + entries = systemPrompt + promptEntries, budget = TokenBudget(limit = stageConfig.tokenBudget), ) @@ -151,11 +181,34 @@ abstract class SessionOrchestrator( is InferenceResult.Success -> { if (isCancelled(sessionId)) return StageExecutionResult.Failure("CANCELLED", retryable = false) val validationCtx = ValidationContext(graph = graph, sessionState = session.state) - mapValidationOutcome(sessionId, stageId, validationCtx) + when (val outcome = mapValidationOutcome(sessionId, stageId, validationCtx)) { + is StageExecutionResult.Success -> verifyProduces(sessionId, stageId, stageConfig) + is StageExecutionResult.Failure -> outcome + } } } } + private fun verifyProduces( + sessionId: SessionId, + stageId: StageId, + stageConfig: StageConfig, + ): StageExecutionResult { + if (stageConfig.produces.isEmpty()) return StageExecutionResult.Success(emptyList()) + val present = artifactRepository.getByIds(sessionId, stageConfig.produces).keys + val missing = stageConfig.produces - present + if (missing.isEmpty()) return StageExecutionResult.Success(emptyList()) + val missingIds = missing.joinToString(", ") { it.value } + log.error( + "[Orchestrator] stage missing produced artifacts " + + "session=${sessionId.value} stage=${stageId.value} missing=$missingIds", + ) + return StageExecutionResult.Failure( + "stage ${stageId.value} did not produce declared artifacts: $missingIds", + retryable = true, + ) + } + internal open suspend fun mapValidationOutcome( sessionId: SessionId, stageId: StageId, @@ -164,6 +217,7 @@ abstract class SessionOrchestrator( is ValidationOutcome.Passed -> StageExecutionResult.Success(emptyList()) is ValidationOutcome.Rejected -> StageExecutionResult.Failure("validation failed", retryable = outcome.retryable) + is ValidationOutcome.NeedsApproval -> handleApproval(sessionId, stageId, outcome) } @@ -177,7 +231,7 @@ abstract class SessionOrchestrator( val provider = inferenceRouter.route(stageId, stageConfig.requiredCapabilities) System.err.println( "[Orchestrator] inference session=${sessionId.value} stage=${stageId.value} " + - "provider=${provider.id.value} timeoutMs=$timeoutMs" + "provider=${provider.id.value} timeoutMs=$timeoutMs", ) val requestId = InferenceRequestId(UUID.randomUUID().toString()) val request = InferenceRequest( @@ -188,7 +242,9 @@ abstract class SessionOrchestrator( generationConfig = stageConfig.generationConfig, ) - emit(sessionId, InferenceStartedEvent(requestId, sessionId, stageId, provider.id)) + // TODO: capture rendered prompt; currently contextpack.toString() + val promptArtifactId = artifactStore.put(contextPack.toString().toByteArray()) + emit(sessionId, InferenceStartedEvent(requestId, sessionId, stageId, provider.id, promptArtifactId)) if (isCancelled(sessionId)) return InferenceResult.Cancelled @@ -196,11 +252,12 @@ abstract class SessionOrchestrator( val response = withTimeout(timeoutMs) { provider.infer(request) } System.err.println( "[Orchestrator] inference done session=${sessionId.value} stage=${stageId.value} " + - "tokens=${response.tokensUsed} latencyMs=${response.latencyMs}" + "tokens=${response.tokensUsed} latencyMs=${response.latencyMs}", ) if (isCancelled(sessionId)) return InferenceResult.Cancelled + val responseArtifactId = artifactStore.put(response.text.toByteArray()) emit( sessionId, InferenceCompletedEvent( @@ -210,6 +267,7 @@ abstract class SessionOrchestrator( providerId = provider.id, tokensUsed = response.tokensUsed, latencyMs = response.latencyMs, + responseArtifactId = responseArtifactId, ), ) @@ -258,7 +316,7 @@ abstract class SessionOrchestrator( internal fun isTerminal(graph: WorkflowGraph, stageId: StageId): Boolean = graph.transitions.none { it.from == stageId } - internal fun advanceStage( + internal suspend fun advanceStage( sessionId: SessionId, fromStageId: StageId, decision: TransitionDecision.Move, @@ -269,7 +327,7 @@ abstract class SessionOrchestrator( // --- terminal state helpers --- - internal fun emitWorkflowStarted(sessionId: SessionId, graph: WorkflowGraph, config: OrchestrationConfig) { + internal suspend fun emitWorkflowStarted(sessionId: SessionId, graph: WorkflowGraph, config: OrchestrationConfig) { emit( sessionId, WorkflowStartedEvent( @@ -280,21 +338,21 @@ abstract class SessionOrchestrator( ) } - internal fun completeWorkflow( + internal suspend fun completeWorkflow( sessionId: SessionId, terminalStageId: StageId, stageCount: Int, ): WorkflowResult.Completed { System.err.println( "[Orchestrator] COMPLETED session=${sessionId.value} " + - "terminalStage=${terminalStageId.value} stages=$stageCount" + "terminalStage=${terminalStageId.value} stages=$stageCount", ) emit(sessionId, WorkflowCompletedEvent(sessionId, terminalStageId, stageCount)) cancellations.remove(sessionId) return WorkflowResult.Completed(sessionId, terminalStageId) } - internal fun failWorkflow( + internal suspend fun failWorkflow( sessionId: SessionId, stageId: StageId, reason: String, @@ -302,14 +360,14 @@ abstract class SessionOrchestrator( ): WorkflowResult.Failed { System.err.println( "[Orchestrator] FAILED session=${sessionId.value} stage=${stageId.value} " + - "reason=$reason retryExhausted=$retryExhausted" + "reason=$reason retryExhausted=$retryExhausted", ) emit(sessionId, WorkflowFailedEvent(sessionId, stageId, reason, retryExhausted)) cancellations.remove(sessionId) return WorkflowResult.Failed(sessionId, reason, retryExhausted) } - internal fun handleCancellation( + internal suspend fun handleCancellation( sessionId: SessionId, stageId: StageId, ): WorkflowResult.Cancelled { @@ -326,7 +384,7 @@ abstract class SessionOrchestrator( // --- event emission --- - internal fun emit(sessionId: SessionId, payload: EventPayload) { + internal suspend fun emit(sessionId: SessionId, payload: EventPayload) { eventStore.append( NewEvent( metadata = EventMetadata( @@ -372,12 +430,12 @@ abstract class SessionOrchestrator( return try { System.err.println( "[Orchestrator] awaiting approval session=${sessionId.value} " + - "requestId=${domainRequest.id.value}" + "requestId=${domainRequest.id.value}", ) val decision = deferred.await() System.err.println( "[Orchestrator] approval resolved session=${sessionId.value} " + - "approved=${decision.isApproved}" + "approved=${decision.isApproved}", ) emitDecisionResolved(sessionId, domainRequest, decision) if (decision.isApproved) { @@ -391,7 +449,7 @@ abstract class SessionOrchestrator( } } - private fun buildApprovalRequest( + private suspend fun buildApprovalRequest( sessionId: SessionId, stageId: StageId, outcome: ValidationOutcome.NeedsApproval, @@ -419,7 +477,7 @@ abstract class SessionOrchestrator( ) } - private fun emitDecisionResolved( + private suspend fun emitDecisionResolved( sessionId: SessionId, domainRequest: DomainApprovalRequest, decision: ApprovalDecision, diff --git a/docs/reviews/2026-05-18-cas-steps-1-8-review.md b/docs/reviews/2026-05-18-cas-steps-1-8-review.md new file mode 100644 index 00000000..c56fb65e --- /dev/null +++ b/docs/reviews/2026-05-18-cas-steps-1-8-review.md @@ -0,0 +1,62 @@ +# CAS Artifact Store — Steps 1–8 Final Review + +**Plan:** `docs/plans/2026-05-17-cas-artifact-store.md` +**Date:** 2026-05-18 +**Scope:** Steps 1–8 (step 9 cloud sync deferred to a later epic) +**Build status:** `:infrastructure:artifacts-cas:check` green, 21/21 tests, detekt + kover clean + +## Summary + +Content-Addressed Store added as a new core interface (`core:artifacts-store`) plus a SQLite+segment-file implementation (`infrastructure/artifacts-cas`). Inference events now carry `promptArtifactId` / `responseArtifactId`; orchestrators `put(...)` bytes before emitting events. `SqliteEventStore.append` wraps its txn in `artifactStore.flushBefore { ... }` so segment data is fsynced before the event commit. Recovery tail-scan re-indexes orphan tail records on reopen, making the crash window between artifact write and event commit non-corrupting. A manual compactor + oldest-first disk-cap evictor round out the lifecycle; both gated by a `maintenanceMutex` so they cannot race each other. + +## CLAUDE.md invariants + +All hold: + +- Event log remains sole source of truth; segment + index are derivable via `TailScanner.recover()`. +- No new `EventPayload` subclasses introduced; `InferenceStartedEvent` / `InferenceCompletedEvent` already registered in `core/events/.../serialization/Serialization.kt`. +- No cross-core impl deps. `core:artifacts-store` depends only on `core:events` (for the `ArtifactId = TypeId` typealias). +- Apps → core → infrastructure DAG preserved. +- Polymorphic registration verified for the modified inference event classes. + +## Findings + +### MUST FIX +None. + +### SHOULD FIX + +1. **`CasArtifactStore.open()` discards the `TailScanReport`** (`infrastructure/artifacts-cas/.../CasArtifactStore.kt:87-91`). If recovery truncates a corrupt tail or re-indexes orphan records, the caller has no signal. Log at INFO when `truncatedAt != null` or `recoveredRecords > 0`. + +2. **No end-to-end orchestrator→put→append→replay→get test.** `SessionOrchestratorIntegrationTest` uses `NoopArtifactStore` plus a `RecordingArtifactStore` to assert put-then-emit ordering, but nothing wires the real `CasArtifactStore` + `SqliteEventStore` together to verify the crash-window invariant in situ. Add one test: put via real CAS, append an inference event, reopen, replay, assert `get(responseArtifactId)` returns the original bytes. + +3. **`SqliteEventStore.allEvents()` (new in this diff) lacks `withContext(Dispatchers.IO)`.** Called from `LivenessScanner` which already wraps the iteration, so currently safe — but for consistency the read methods on the store itself should dispatch internally. + +4. **`ReplayOrchestrator.runInference` super path still calls `artifactStore.put`** for the prompt before checking strategy (`core/kernel/.../ReplayOrchestrator.kt:178`). For non-`SkipInference` strategies during replay this writes a new artifact for an event being re-derived, bloating the store with duplicates. Verify intent; if unintentional, gate `put` on a replay flag. + +### NICE TO HAVE + +- `SegmentWriter.append` does not fsync per-record; durability comes from `flushBefore`. Document this contract in `ArtifactStore.kt` KDoc. +- `Compactor.readLiveEntries` builds `liveHashHex` from `liveIds.value` (hex) and compares to `hash.toHex()` per entry. A `ByteArrayKey` wrapper would avoid per-entry hex allocation. +- `CasArtifactStore.evict` does `listDirectoryEntries` under `mutex` to determine `activeId`. `SegmentWriter` already tracks `currentSegmentId` — expose a getter and skip the filesystem call. + +## Categories with no findings + +- Polymorphic serialization registration +- Cross-core deps +- Resource hygiene (`AutoCloseable`, `.use { }` chains) +- Detekt antipatterns (`Thread.sleep`, bare try/catch, new `@Suppress`) + +## Per-step inventory + +| Step | Subject | Status | +|------|----------------------------------------------------------------------|--------| +| 1 | `ArtifactId` value class in `core:events` | reused existing `TypeId` typealias | +| 2 | `core:artifacts-store` interface module | done | +| 3 | `infrastructure/artifacts-cas` MVP (writer/reader/index/store) | done | +| 4 | DI wiring + `flushBefore` around SQLite commit | done | +| 5 | Event schema (`promptArtifactId` / `responseArtifactId`) | done | +| 6 | Recovery tail-scan | done | +| 7 | Compactor + LivenessScanner | done | +| 8 | Disk-cap eviction policy | done | +| 9 | Cloud sync | deferred (later epic) | diff --git a/infrastructure/artifacts-cas/build.gradle b/infrastructure/artifacts-cas/build.gradle new file mode 100644 index 00000000..b75ae76a --- /dev/null +++ b/infrastructure/artifacts-cas/build.gradle @@ -0,0 +1,16 @@ +plugins { + id 'java-library' + id 'org.jetbrains.kotlin.jvm' +} + +dependencies { + implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version" + implementation(project(":core:events")) + implementation(project(":core:artifacts-store")) + implementation "org.xerial:sqlite-jdbc" + implementation "org.bouncycastle:bcprov-jdk18on:1.78.1" + testImplementation "org.junit.jupiter:junit-jupiter" + testImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-test:$kotlinx_coroutines_version" + testImplementation(project(":infrastructure:persistence")) + testImplementation(testFixtures(project(":testing:contracts"))) +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStore.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStore.kt new file mode 100644 index 00000000..81ec9451 --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStore.kt @@ -0,0 +1,108 @@ +package com.correx.infrastructure.artifactscas + +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.ArtifactId +import com.correx.core.utils.TypeId +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.index.ArtifactIndex +import com.correx.infrastructure.artifactscas.recovery.TailScanReport +import com.correx.infrastructure.artifactscas.recovery.TailScanner +import com.correx.infrastructure.artifactscas.segment.SegmentLayout +import com.correx.infrastructure.artifactscas.segment.SegmentReader +import com.correx.infrastructure.artifactscas.segment.SegmentWriter +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import java.nio.file.Files +import kotlin.io.path.listDirectoryEntries + +class CasArtifactStore( + private val config: CasConfig, + private val index: ArtifactIndex, +) : ArtifactStore, AutoCloseable { + private val segmentsDir = config.rootDir.resolve("segments").also { Files.createDirectories(it) } + private val writer = SegmentWriter(segmentsDir, config.maxSegmentBytes) + private val reader = SegmentReader() + private val mutex = Mutex() + private val maintenanceMutex = Mutex() + + override suspend fun put(bytes: ByteArray): ArtifactId { + val hash = SegmentLayout.blake3(bytes) + return mutex.withLock { + val existing = index.lookup(hash) + if (existing == null) { + val loc = writer.append(bytes, hash) + index.insert(hash, loc) + } + TypeId(hash.toHex()) + } + } + + override suspend fun get(id: ArtifactId): ByteArray? { + val hash = id.value.hexToBytes() + val loc = mutex.withLock { index.lookup(hash) } ?: return null + return reader.read(loc, segmentsDir) + } + + override suspend fun flushBefore(commit: suspend () -> Unit) { + mutex.withLock { + writer.fsync() + index.flush() + commit() + } + } + + override fun close() { + writer.close() + index.close() + } + + suspend fun recover(): TailScanReport? = TailScanner(segmentsDir, index).recover() + + fun compactor(eventStore: EventStore): Compactor = + Compactor(config, index, LivenessScanner(eventStore, index), storeMutex = mutex) + + // Run synchronously; safe to call alongside puts (acquires storeMutex briefly when swapping segments). + // maintenanceMutex makes compact and evict mutually exclusive so they cannot race on liveness/segment files. + suspend fun compact(eventStore: EventStore): CompactionReport = + maintenanceMutex.withLock { compactor(eventStore).compact() } + + suspend fun evict(eventStore: EventStore): EvictionReport = + maintenanceMutex.withLock { + val evictor = Evictor(config, index, LivenessScanner(eventStore, index)) + mutex.withLock { + val activeId = segmentsDir.listDirectoryEntries("*.seg") + .mapNotNull { runCatching { it.fileName.toString().removeSuffix(".seg").toLong() }.getOrNull() } + .maxOrNull() ?: return@withLock EvictionReport(emptyList(), 0L) + evictor.evict(activeId) + } + } + + companion object { + suspend fun open(config: CasConfig, index: ArtifactIndex): CasArtifactStore { + val segmentsDir = config.rootDir.resolve("segments").also { Files.createDirectories(it) } + TailScanner(segmentsDir, index).recover() + return CasArtifactStore(config, index) + } + } +} + +private fun ByteArray.toHex(): String = joinToString("") { "%02x".format(it) } + +private const val HEX_RADIX = 16 +private const val HEX_CHARS_PER_BYTE = 2 +private const val HEX_NIBBLE_BITS = 4 + +private fun String.hexToBytes(): ByteArray { + require(length % HEX_CHARS_PER_BYTE == 0) { "hex string must have even length" } + return ByteArray(length / HEX_CHARS_PER_BYTE) { i -> + val hi = Character.digit(this[HEX_CHARS_PER_BYTE * i], HEX_RADIX) + val lo = Character.digit(this[HEX_CHARS_PER_BYTE * i + 1], HEX_RADIX) + ((hi shl HEX_NIBBLE_BITS) + lo).toByte() + } +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/Compactor.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/Compactor.kt new file mode 100644 index 00000000..a869c0fd --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/Compactor.kt @@ -0,0 +1,146 @@ +package com.correx.infrastructure.artifactscas.compact + +import com.correx.infrastructure.artifactscas.config.CasConfig +import com.correx.infrastructure.artifactscas.index.ArtifactIndex +import com.correx.infrastructure.artifactscas.segment.Location +import com.correx.infrastructure.artifactscas.segment.SegmentLayout +import com.correx.infrastructure.artifactscas.segment.SegmentReader +import com.correx.infrastructure.artifactscas.segment.SegmentWriter +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext +import java.io.RandomAccessFile +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries + +data class CompactionReport( + val compactedSegmentIds: List, + val newSegmentId: Long?, + val bytesReclaimed: Long, +) + +class Compactor( + config: CasConfig, + private val index: ArtifactIndex, + private val liveness: LivenessScanner, + private val storeMutex: Mutex, + private val liveRatioThreshold: Double = DEFAULT_LIVE_RATIO_THRESHOLD, +) { + private val segmentsDir: Path = config.rootDir.resolve("segments") + private val reader = SegmentReader() + + suspend fun compact(): CompactionReport = withContext(Dispatchers.IO) { + if (!Files.exists(segmentsDir)) return@withContext empty() + val segmentIds = listSegmentIds() + if (segmentIds.isEmpty()) return@withContext empty() + val activeId = segmentIds.max() + val scan = liveness.scan(segmentsDir) + val eligible = selectEligible(scan, activeId) + if (eligible.isEmpty()) return@withContext empty() + + val liveEntries = readLiveEntries(eligible, scan.liveIds) + val newSegmentId = activeId + 1 + val (bytesWritten, newLocations) = writeNewSegment(newSegmentId, liveEntries) + + val reclaimedBefore = eligible.sumOf { id -> scan.perSegment[id]?.totalBytes ?: 0L } + applyAtomicSwap(newLocations, eligible) + + val effectiveNew = if (liveEntries.isEmpty()) { + runCatching { Files.deleteIfExists(segmentsDir.resolve(SegmentWriter.segmentFileName(newSegmentId))) } + null + } else { + newSegmentId + } + + CompactionReport( + compactedSegmentIds = eligible, + newSegmentId = effectiveNew, + bytesReclaimed = reclaimedBefore - bytesWritten, + ) + } + + private fun selectEligible(scan: LivenessScanner.Liveness, activeId: Long): List = + scan.perSegment + .filter { (id, stats) -> + id != activeId && stats.totalBytes > 0 && stats.liveRatio < liveRatioThreshold + } + .keys + .sorted() + + private suspend fun readLiveEntries( + eligible: List, + liveIds: Set, + ): List { + val liveHashHex = liveIds.map { it.value }.toHashSet() + val out = mutableListOf() + for (segId in eligible) { + for ((hash, loc) in index.entriesIn(segId)) { + if (hash.toHex() in liveHashHex) { + val bytes = reader.read(loc, segmentsDir) + out.add(LiveEntry(hash, bytes)) + } + } + } + return out + } + + private suspend fun writeNewSegment( + newSegmentId: Long, + entries: List, + ): Pair>> { + if (entries.isEmpty()) return 0L to emptyList() + val newPath = segmentsDir.resolve(SegmentWriter.segmentFileName(newSegmentId)) + var bytesWritten = 0L + val newLocations = mutableListOf>() + withContext(Dispatchers.IO) { + RandomAccessFile(newPath.toFile(), "rw").use { raf -> + for (entry in entries) { + val header = SegmentLayout.encodeHeader(entry.payload, entry.hash) + val offset = raf.filePointer + raf.write(header) + raf.write(entry.payload) + newLocations.add(entry.hash to Location(newSegmentId, offset, entry.payload.size)) + bytesWritten += header.size + entry.payload.size + } + raf.fd.sync() + } + } + return bytesWritten to newLocations + } + + private suspend fun applyAtomicSwap(newLocations: List>, eligible: List) { + storeMutex.withLock { + for ((hash, loc) in newLocations) { + index.update(hash, loc) + } + for (segId in eligible) { + runCatching { + Files.deleteIfExists(segmentsDir.resolve(SegmentWriter.segmentFileName(segId))) + } + } + for (segId in eligible) { + index.deleteEntriesIn(segId) + } + index.flush() + } + } + + private fun listSegmentIds(): List = + segmentsDir.listDirectoryEntries("*.seg") + .mapNotNull { p -> runCatching { p.fileName.toString().removeSuffix(".seg").toLong() }.getOrNull() } + + private fun empty(): CompactionReport = CompactionReport(emptyList(), null, 0L) + + private class LiveEntry( + val hash: ByteArray, + val payload: ByteArray, + ) + + companion object { + const val DEFAULT_LIVE_RATIO_THRESHOLD = 0.5 + } +} + +private fun ByteArray.toHex(): String = joinToString("") { "%02x".format(it) } diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/LivenessScanner.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/LivenessScanner.kt new file mode 100644 index 00000000..fc892f15 --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/compact/LivenessScanner.kt @@ -0,0 +1,77 @@ +package com.correx.infrastructure.artifactscas.compact + +import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.InferenceStartedEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.ArtifactId +import com.correx.infrastructure.artifactscas.index.ArtifactIndex +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries + +class LivenessScanner( + private val eventStore: EventStore, + private val index: ArtifactIndex, +) { + data class SegmentStats(val liveBytes: Long, val totalBytes: Long) { + val liveRatio: Double get() = if (totalBytes == 0L) 1.0 else liveBytes.toDouble() / totalBytes + } + + data class Liveness( + val liveIds: Set, + val perSegment: Map, + ) + + suspend fun scan(): Liveness = scan(segmentsDir = null) + + suspend fun scan(segmentsDir: Path?): Liveness { + val liveIds = collectLiveIds() + val perSegmentLive = mutableMapOf() + for (id in liveIds) { + val hash = id.value.hexToBytes() + val loc = index.lookup(hash) ?: continue + perSegmentLive.merge(loc.segmentId, loc.length.toLong()) { a, b -> a + b } + } + val perSegment = if (segmentsDir != null && Files.exists(segmentsDir)) { + segmentsDir.listDirectoryEntries("*.seg") + .mapNotNull { p -> + val id = runCatching { p.fileName.toString().removeSuffix(".seg").toLong() }.getOrNull() + ?: return@mapNotNull null + val total = runCatching { Files.size(p) }.getOrDefault(0L) + val live = perSegmentLive[id] ?: 0L + id to SegmentStats(live, total) + }.toMap() + } else { + perSegmentLive.mapValues { (_, live) -> SegmentStats(live, live) } + } + return Liveness(liveIds, perSegment) + } + + private suspend fun collectLiveIds(): Set = withContext(Dispatchers.IO) { + val ids = mutableSetOf() + for (event in eventStore.allEvents()) { + when (val p = event.payload) { + is InferenceStartedEvent -> ids.add(p.promptArtifactId) + is InferenceCompletedEvent -> ids.add(p.responseArtifactId) + else -> Unit + } + } + ids + } + +} + +private const val HEX_RADIX = 16 +private const val HEX_CHARS_PER_BYTE = 2 +private const val HEX_NIBBLE_BITS = 4 + +internal fun String.hexToBytes(): ByteArray { + require(length % HEX_CHARS_PER_BYTE == 0) { "hex string must have even length" } + return ByteArray(length / HEX_CHARS_PER_BYTE) { i -> + val hi = Character.digit(this[HEX_CHARS_PER_BYTE * i], HEX_RADIX) + val lo = Character.digit(this[HEX_CHARS_PER_BYTE * i + 1], HEX_RADIX) + ((hi shl HEX_NIBBLE_BITS) + lo).toByte() + } +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/config/CasConfig.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/config/CasConfig.kt new file mode 100644 index 00000000..af80b93e --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/config/CasConfig.kt @@ -0,0 +1,14 @@ +package com.correx.infrastructure.artifactscas.config + +import java.nio.file.Path + +data class CasConfig( + val rootDir: Path, + val maxSegmentBytes: Long = DEFAULT_MAX_SEGMENT_BYTES, + val maxTotalBytes: Long = DEFAULT_MAX_TOTAL_BYTES, +) { + companion object { + const val DEFAULT_MAX_SEGMENT_BYTES: Long = 64L * 1024 * 1024 + const val DEFAULT_MAX_TOTAL_BYTES: Long = Long.MAX_VALUE + } +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/evict/Evictor.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/evict/Evictor.kt new file mode 100644 index 00000000..0d105e6a --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/evict/Evictor.kt @@ -0,0 +1,61 @@ +package com.correx.infrastructure.artifactscas.evict + +import com.correx.infrastructure.artifactscas.compact.LivenessScanner +import com.correx.infrastructure.artifactscas.config.CasConfig +import com.correx.infrastructure.artifactscas.index.ArtifactIndex +import com.correx.infrastructure.artifactscas.segment.SegmentWriter +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries + +data class EvictionReport( + val evictedSegmentIds: List, + val bytesReclaimed: Long, +) + +class Evictor( + config: CasConfig, + private val index: ArtifactIndex, + private val liveness: LivenessScanner, +) { + private val segmentsDir: Path = config.rootDir.resolve("segments") + private val maxTotalBytes: Long = config.maxTotalBytes + + suspend fun evict(activeSegmentId: Long): EvictionReport = + withContext(Dispatchers.IO) { + if (!Files.exists(segmentsDir)) return@withContext empty() + val sizes = listSegmentSizes() + if (sizes.isEmpty()) return@withContext empty() + var totalBytes = sizes.values.sum() + if (totalBytes <= maxTotalBytes) return@withContext empty() + + val scan = liveness.scan(segmentsDir) + val evicted = mutableListOf() + var reclaimed = 0L + val candidates = sizes.keys.sorted() + .filter { it != activeSegmentId && (scan.perSegment[it]?.liveBytes ?: 0L) == 0L } + for (id in candidates) { + if (totalBytes <= maxTotalBytes) break + val size = sizes.getValue(id) + index.deleteEntriesIn(id) + runCatching { Files.deleteIfExists(segmentsDir.resolve(SegmentWriter.segmentFileName(id))) } + evicted.add(id) + reclaimed += size + totalBytes -= size + } + if (evicted.isEmpty()) empty() else EvictionReport(evicted, reclaimed) + } + + private fun listSegmentSizes(): Map = + segmentsDir.listDirectoryEntries("*.seg") + .mapNotNull { p -> + val id = runCatching { p.fileName.toString().removeSuffix(".seg").toLong() }.getOrNull() + ?: return@mapNotNull null + val size = runCatching { Files.size(p) }.getOrDefault(0L) + id to size + }.toMap() + + private fun empty(): EvictionReport = EvictionReport(emptyList(), 0L) +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/ArtifactIndex.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/ArtifactIndex.kt new file mode 100644 index 00000000..be424e2b --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/ArtifactIndex.kt @@ -0,0 +1,13 @@ +package com.correx.infrastructure.artifactscas.index + +import com.correx.infrastructure.artifactscas.segment.Location + +interface ArtifactIndex : AutoCloseable { + suspend fun lookup(hash: ByteArray): Location? + suspend fun insert(hash: ByteArray, loc: Location) + suspend fun update(hash: ByteArray, loc: Location) + suspend fun entriesIn(segmentId: Long): List> + suspend fun deleteEntriesIn(segmentId: Long) + suspend fun flush() + suspend fun maxKnownTailFor(segmentId: Long): Long +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/SqliteArtifactIndex.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/SqliteArtifactIndex.kt new file mode 100644 index 00000000..c8d68efe --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/index/SqliteArtifactIndex.kt @@ -0,0 +1,136 @@ +package com.correx.infrastructure.artifactscas.index + +import com.correx.infrastructure.artifactscas.segment.Location +import com.correx.infrastructure.artifactscas.segment.SegmentLayout +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.nio.file.Path +import java.sql.Connection +import java.sql.DriverManager + +class SqliteArtifactIndex(dbPath: Path) : ArtifactIndex { + private val connection: Connection = DriverManager.getConnection("jdbc:sqlite:${dbPath.toAbsolutePath()}") + + init { + connection.createStatement().use { stmt -> + stmt.execute("PRAGMA journal_mode=WAL;") + stmt.execute("PRAGMA synchronous=NORMAL;") + stmt.execute( + """ + CREATE TABLE IF NOT EXISTS artifacts ( + hash BLOB PRIMARY KEY, + segment_id INTEGER NOT NULL, + offset INTEGER NOT NULL, + length INTEGER NOT NULL + ); + """.trimIndent() + ) + } + } + + override suspend fun lookup(hash: ByteArray): Location? = withContext(Dispatchers.IO) { + connection.prepareStatement( + "SELECT segment_id, offset, length FROM artifacts WHERE hash = ?" + ).use { ps -> + ps.setBytes(PARAM_HASH, hash) + ps.executeQuery().use { rs -> + if (!rs.next()) { + null + } else { + Location(rs.getLong(COL_SEGMENT_ID), rs.getLong(COL_OFFSET), rs.getInt(COL_LENGTH)) + } + } + } + } + + override suspend fun insert(hash: ByteArray, loc: Location): Unit = withContext(Dispatchers.IO) { + // Same hash always maps to same location; duplicate inserts are no-ops. + connection.prepareStatement( + "INSERT OR IGNORE INTO artifacts (hash, segment_id, offset, length) VALUES (?, ?, ?, ?)" + ).use { ps -> + ps.setBytes(PARAM_HASH, hash) + ps.setLong(PARAM_SEGMENT_ID, loc.segmentId) + ps.setLong(PARAM_OFFSET, loc.offset) + ps.setInt(PARAM_LENGTH, loc.length) + ps.executeUpdate() + } + } + + override suspend fun update(hash: ByteArray, loc: Location): Unit = withContext(Dispatchers.IO) { + connection.prepareStatement( + """ + INSERT INTO artifacts (hash, segment_id, offset, length) VALUES (?, ?, ?, ?) + ON CONFLICT(hash) DO UPDATE SET + segment_id = excluded.segment_id, + offset = excluded.offset, + length = excluded.length + """.trimIndent() + ).use { ps -> + ps.setBytes(PARAM_HASH, hash) + ps.setLong(PARAM_SEGMENT_ID, loc.segmentId) + ps.setLong(PARAM_OFFSET, loc.offset) + ps.setInt(PARAM_LENGTH, loc.length) + ps.executeUpdate() + } + } + + override suspend fun entriesIn(segmentId: Long): List> = withContext(Dispatchers.IO) { + connection.prepareStatement( + "SELECT hash, offset, length FROM artifacts WHERE segment_id = ? ORDER BY offset ASC" + ).use { ps -> + ps.setLong(1, segmentId) + ps.executeQuery().use { rs -> + buildList { + while (rs.next()) { + val hash = rs.getBytes(COL_ENTRY_HASH) + val loc = Location(segmentId, rs.getLong(COL_ENTRY_OFFSET), rs.getInt(COL_ENTRY_LENGTH)) + add(hash to loc) + } + } + } + } + } + + override suspend fun deleteEntriesIn(segmentId: Long): Unit = withContext(Dispatchers.IO) { + connection.prepareStatement("DELETE FROM artifacts WHERE segment_id = ?").use { ps -> + ps.setLong(1, segmentId) + ps.executeUpdate() + } + } + + override suspend fun maxKnownTailFor(segmentId: Long): Long = withContext(Dispatchers.IO) { + connection.prepareStatement( + "SELECT offset, length FROM artifacts WHERE segment_id = ? ORDER BY offset DESC LIMIT 1" + ).use { ps -> + ps.setLong(1, segmentId) + ps.executeQuery().use { rs -> + if (rs.next()) { + rs.getLong(1) + SegmentLayout.HEADER_SIZE + rs.getInt(2) + } else { + 0L + } + } + } + } + + override suspend fun flush(): Unit = withContext(Dispatchers.IO) { + connection.createStatement().use { it.execute("PRAGMA wal_checkpoint(TRUNCATE);") } + } + + override fun close() { + connection.close() + } + + private companion object { + const val PARAM_HASH = 1 + const val PARAM_SEGMENT_ID = 2 + const val PARAM_OFFSET = 3 + const val PARAM_LENGTH = 4 + const val COL_SEGMENT_ID = 1 + const val COL_OFFSET = 2 + const val COL_LENGTH = 3 + const val COL_ENTRY_HASH = 1 + const val COL_ENTRY_OFFSET = 2 + const val COL_ENTRY_LENGTH = 3 + } +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScanner.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScanner.kt new file mode 100644 index 00000000..a554c6d6 --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScanner.kt @@ -0,0 +1,75 @@ +package com.correx.infrastructure.artifactscas.recovery + +import com.correx.infrastructure.artifactscas.index.ArtifactIndex +import com.correx.infrastructure.artifactscas.segment.Location +import com.correx.infrastructure.artifactscas.segment.SegmentLayout +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.io.RandomAccessFile +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries + +data class TailScanReport( + val activeSegmentId: Long, + val knownTailOffset: Long, + val recoveredRecords: Int, + val truncatedAt: Long?, +) + +class TailScanner( + private val segmentsDir: Path, + private val index: ArtifactIndex, +) { + suspend fun recover(): TailScanReport? = withContext(Dispatchers.IO) { + if (!Files.exists(segmentsDir)) return@withContext null + val active = segmentsDir.listDirectoryEntries("*.seg") + .mapNotNull { p -> + runCatching { p.fileName.toString().removeSuffix(".seg").toLong() to p }.getOrNull() + } + .maxByOrNull { it.first } + ?: return@withContext null + val (activeId, activePath) = active + val knownTail = index.maxKnownTailFor(activeId) + scanAndRecover(activeId, activePath, knownTail) + } + + private suspend fun scanAndRecover(activeId: Long, path: Path, startOffset: Long): TailScanReport { + var recovered = 0 + var truncatedAt: Long? = null + RandomAccessFile(path.toFile(), "rw").use { raf -> + val fileLen = raf.length() + var pos = startOffset + while (pos < fileLen) { + val step = tryReadRecord(raf, pos, fileLen) + if (step == null) { + truncatedAt = pos + break + } + index.insert(step.hash, Location(activeId, pos, step.length)) + recovered += 1 + pos += SegmentLayout.HEADER_SIZE + step.length + } + // Truncate at first bad/partial record so writer resumes at a clean boundary. + truncatedAt?.let { raf.setLength(it) } + } + return TailScanReport(activeId, startOffset, recovered, truncatedAt) + } + + private fun tryReadRecord(raf: RandomAccessFile, pos: Long, fileLen: Long): ValidRecord? = + runCatching { + require(fileLen - pos >= SegmentLayout.HEADER_SIZE) + raf.seek(pos) + val header = ByteArray(SegmentLayout.HEADER_SIZE) + raf.readFully(header) + val (length, crc, hash) = SegmentLayout.decodeHeader(header) + require(length >= 0 && pos + SegmentLayout.HEADER_SIZE + length <= fileLen) + val payload = ByteArray(length) + raf.readFully(payload) + require(SegmentLayout.crc32c(payload) == crc) + require(SegmentLayout.blake3(payload).contentEquals(hash)) + ValidRecord(length, hash) + }.getOrNull() + + private class ValidRecord(val length: Int, val hash: ByteArray) +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentLayout.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentLayout.kt new file mode 100644 index 00000000..9c51a426 --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentLayout.kt @@ -0,0 +1,44 @@ +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 + +object SegmentLayout { + const val HASH_SIZE: Int = 32 + const val HEADER_SIZE: Int = 4 + 4 + HASH_SIZE + private const val BLAKE3_DIGEST_BITS: Int = 256 + + fun encodeHeader(payload: ByteArray, hash: ByteArray): ByteArray { + require(hash.size == HASH_SIZE) { "hash must be $HASH_SIZE bytes" } + val crc = CRC32C().apply { update(payload, 0, payload.size) }.value.toInt() + return ByteBuffer.allocate(HEADER_SIZE) + .order(ByteOrder.BIG_ENDIAN) + .putInt(payload.size) + .putInt(crc) + .put(hash) + .array() + } + + fun decodeHeader(header: ByteArray): Triple { + require(header.size == HEADER_SIZE) { "header must be $HEADER_SIZE bytes" } + val buf = ByteBuffer.wrap(header).order(ByteOrder.BIG_ENDIAN) + val length = buf.int + val crc = buf.int + val hash = ByteArray(HASH_SIZE) + buf.get(hash) + return Triple(length, crc, hash) + } + + fun blake3(bytes: ByteArray): ByteArray { + val digest = Blake3Digest(BLAKE3_DIGEST_BITS) + digest.update(bytes, 0, bytes.size) + val out = ByteArray(HASH_SIZE) + digest.doFinal(out, 0) + return out + } + + fun crc32c(payload: ByteArray): Int = + CRC32C().apply { update(payload, 0, payload.size) }.value.toInt() +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentReader.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentReader.kt new file mode 100644 index 00000000..7f031d96 --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentReader.kt @@ -0,0 +1,32 @@ +package com.correx.infrastructure.artifactscas.segment + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.io.RandomAccessFile +import java.nio.file.Path + +class CorruptRecordException(message: String) : RuntimeException(message) + +class SegmentReader { + suspend fun read(loc: Location, segmentsDir: Path): ByteArray = withContext(Dispatchers.IO) { + val path = segmentsDir.resolve(SegmentWriter.segmentFileName(loc.segmentId)) + RandomAccessFile(path.toFile(), "r").use { raf -> + raf.seek(loc.offset) + val header = ByteArray(SegmentLayout.HEADER_SIZE) + raf.readFully(header) + val (length, crc, hash) = SegmentLayout.decodeHeader(header) + if (length != loc.length) { + throw CorruptRecordException("length mismatch at seg=${loc.segmentId} off=${loc.offset}") + } + val payload = ByteArray(length) + raf.readFully(payload) + if (SegmentLayout.crc32c(payload) != crc) { + throw CorruptRecordException("crc mismatch at seg=${loc.segmentId} off=${loc.offset}") + } + if (!SegmentLayout.blake3(payload).contentEquals(hash)) { + throw CorruptRecordException("hash mismatch at seg=${loc.segmentId} off=${loc.offset}") + } + payload + } + } +} diff --git a/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentWriter.kt b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentWriter.kt new file mode 100644 index 00000000..b1519816 --- /dev/null +++ b/infrastructure/artifacts-cas/src/main/kotlin/com/correx/infrastructure/artifactscas/segment/SegmentWriter.kt @@ -0,0 +1,74 @@ +package com.correx.infrastructure.artifactscas.segment + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import java.io.RandomAccessFile +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries + +data class Location(val segmentId: Long, val offset: Long, val length: Int) + +class SegmentWriter( + private val segmentsDir: Path, + private val maxSegmentBytes: Long, +) : AutoCloseable { + private var currentSegmentId: Long = 0 + private var currentRaf: RandomAccessFile? = null + private var currentOffset: Long = 0 + + init { + Files.createDirectories(segmentsDir) + openOrCreateActive() + } + + suspend fun append(payload: ByteArray, hash: ByteArray): Location = withContext(Dispatchers.IO) { + rotateIfNeeded(payload.size) + val raf = currentRaf!! + val offset = currentOffset + val header = SegmentLayout.encodeHeader(payload, hash) + raf.write(header) + raf.write(payload) + currentOffset += header.size + payload.size + Location(currentSegmentId, offset, payload.size) + } + + suspend fun fsync() = withContext(Dispatchers.IO) { + currentRaf?.fd?.sync() + Unit + } + + override fun close() { + currentRaf?.close() + currentRaf = null + } + + private fun openOrCreateActive() { + val highest = segmentsDir.listDirectoryEntries("*.seg") + .mapNotNull { runCatching { it.fileName.toString().removeSuffix(".seg").toLong() }.getOrNull() } + .maxOrNull() + currentSegmentId = highest ?: 1L + val path = segmentsDir.resolve(segmentFileName(currentSegmentId)) + val raf = RandomAccessFile(path.toFile(), "rw") + raf.seek(raf.length()) + currentRaf = raf + currentOffset = raf.length() + } + + private fun rotateIfNeeded(incomingPayloadBytes: Int) { + val recordBytes = SegmentLayout.HEADER_SIZE.toLong() + incomingPayloadBytes + if (currentOffset == 0L) return + if (currentOffset + recordBytes <= maxSegmentBytes) return + currentRaf?.fd?.sync() + currentRaf?.close() + currentSegmentId += 1 + val path = segmentsDir.resolve(segmentFileName(currentSegmentId)) + val raf = RandomAccessFile(path.toFile(), "rw") + currentRaf = raf + currentOffset = 0 + } + + companion object { + fun segmentFileName(id: Long): String = "%08d.seg".format(id) + } +} diff --git a/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStoreTest.kt b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStoreTest.kt new file mode 100644 index 00000000..5c4d7b73 --- /dev/null +++ b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/CasArtifactStoreTest.kt @@ -0,0 +1,101 @@ +package com.correx.infrastructure.artifactscas + +import com.correx.core.utils.TypeId +import com.correx.infrastructure.artifactscas.config.CasConfig +import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex +import com.correx.infrastructure.artifactscas.segment.CorruptRecordException +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertArrayEquals +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Assertions.assertThrows +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.io.TempDir +import java.nio.file.Files +import java.nio.file.Path +import kotlin.io.path.listDirectoryEntries +import kotlin.random.Random + +class CasArtifactStoreTest { + + private fun newStore(dir: Path, maxSegmentBytes: Long = 64L * 1024 * 1024): CasArtifactStore { + val index = SqliteArtifactIndex(dir.resolve("index.sqlite")) + return CasArtifactStore(CasConfig(dir, maxSegmentBytes), index) + } + + @Test + fun `roundtrip put then get returns same bytes`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir).use { store -> + val small = "hello cas".toByteArray() + val big = Random(1).nextBytes(1024 * 1024) + val idSmall = store.put(small) + val idBig = store.put(big) + assertArrayEquals(small, store.get(idSmall)) + assertArrayEquals(big, store.get(idBig)) + } + } + + private fun segmentsTotalSize(dir: Path): Long = + Files.walk(dir.resolve("segments")) + .filter { Files.isRegularFile(it) } + .mapToLong { Files.size(it) } + .sum() + + @Test + fun `idempotent put returns same id and does not grow segments`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir).use { store -> + val payload = Random(2).nextBytes(500) + val id1 = store.put(payload) + val sizeAfterFirst = segmentsTotalSize(dir) + val id2 = store.put(payload) + val sizeAfterSecond = segmentsTotalSize(dir) + assertEquals(id1, id2) + assertEquals(sizeAfterFirst, sizeAfterSecond) + } + } + + @Test + fun `get for unknown id returns null`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir).use { store -> + val bogus = TypeId("00".repeat(32)) + assertNull(store.get(bogus)) + } + } + + @Test + fun `restart after fsync reopened store reads previous bytes`(@TempDir dir: Path): Unit = runBlocking { + val payload = "persistent".toByteArray() + val id = newStore(dir).use { store -> + val id = store.put(payload) + store.flushBefore { /* no-op event commit */ } + id + } + newStore(dir).use { reopened -> + assertArrayEquals(payload, reopened.get(id)) + } + } + + @Test + fun `rotation exceeding threshold opens a new segment`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 1024).use { store -> + store.put(Random(3).nextBytes(600)) + store.put(Random(4).nextBytes(600)) + val files = dir.resolve("segments").listDirectoryEntries("*.seg") + assertEquals(2, files.size, "expected rotation, found segments: ${files.map { it.fileName }}") + } + } + + @Test + fun `get throws CorruptRecordException when segment payload is tampered`(@TempDir dir: Path): Unit = runBlocking { + val id = newStore(dir).use { store -> + store.put("tamper me".toByteArray()).also { store.flushBefore {} } + } + val segFile = dir.resolve("segments").listDirectoryEntries("*.seg").single() + val bytes = Files.readAllBytes(segFile) + bytes[bytes.size - 1] = (bytes[bytes.size - 1].toInt() xor 0xFF).toByte() + Files.write(segFile, bytes) + newStore(dir).use { reopened -> + assertThrows(CorruptRecordException::class.java) { runBlocking { reopened.get(id) } } + } + } +} diff --git a/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/compact/CompactorTest.kt b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/compact/CompactorTest.kt new file mode 100644 index 00000000..11bbfa50 --- /dev/null +++ b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/compact/CompactorTest.kt @@ -0,0 +1,200 @@ +package com.correx.infrastructure.artifactscas.compact + +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.InferenceStartedEvent +import com.correx.core.events.events.NewEvent +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.inference.TokenUsage +import com.correx.core.utils.TypeId +import com.correx.infrastructure.artifactscas.CasArtifactStore +import com.correx.infrastructure.artifactscas.config.CasConfig +import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex +import com.correx.infrastructure.persistence.InMemoryEventStore +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertArrayEquals +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 org.junit.jupiter.api.io.TempDir +import java.nio.file.Files +import java.nio.file.Path +import java.util.UUID +import kotlin.io.path.listDirectoryEntries + +class CompactorTest { + + private val sessionId: SessionId = TypeId("session-1") + + private fun newStore(dir: Path, maxSegmentBytes: Long = 64L * 1024 * 1024): CasArtifactStore { + val index = SqliteArtifactIndex(dir.resolve("index.sqlite")) + return CasArtifactStore(CasConfig(dir, maxSegmentBytes), index) + } + + private suspend fun emitStarted(store: InMemoryEventStore, prompt: ArtifactId) { + store.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = InferenceStartedEvent( + requestId = TypeId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = TypeId("stage"), + providerId = TypeId("prov"), + promptArtifactId = prompt, + ), + ) + ) + } + + private suspend fun emitCompleted(store: InMemoryEventStore, response: ArtifactId) { + store.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = InferenceCompletedEvent( + requestId = TypeId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = TypeId("stage"), + providerId = TypeId("prov"), + tokensUsed = TokenUsage(0, 0), + latencyMs = 0L, + responseArtifactId = response, + ), + ) + ) + } + + @Test + fun `no compaction when all artifacts live`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 256).use { store -> + val eventStore = InMemoryEventStore() + val a = store.put("aaaa".toByteArray()) + val b = store.put("bbbb".toByteArray()) + val c = store.put("cccc".toByteArray()) + emitCompleted(eventStore, a) + emitCompleted(eventStore, b) + emitCompleted(eventStore, c) + val report = store.compact(eventStore) + assertTrue(report.compactedSegmentIds.isEmpty(), "report: $report") + } + } + + @Test + fun `compacts a mostly-dead sealed segment`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64).use { store -> + val eventStore = InMemoryEventStore() + val live = store.put(ByteArray(40) { 1 }) + val dead = store.put(ByteArray(40) { 2 }) + store.put(ByteArray(40) { 3 }) + emitCompleted(eventStore, live) + + val segsBefore = dir.resolve("segments").listDirectoryEntries("*.seg") + .map { it.fileName.toString() }.toSet() + assertTrue(segsBefore.size >= 3, "expected rotation, got $segsBefore") + + val report = store.compact(eventStore) + assertFalse(report.compactedSegmentIds.isEmpty(), "expected sealed compactions, got $report") + + assertArrayEquals(ByteArray(40) { 1 }, store.get(live)) + assertNull(store.get(dead)) + } + } + + @Test + fun `fully dead sealed segment is deleted with no new segment written`(@TempDir dir: Path): Unit = runBlocking { + val sealedName: String + val sealedBytes: Long + val sealedSegId: Long + newStore(dir, maxSegmentBytes = 64).use { store -> + val eventStore = InMemoryEventStore() + val dead = store.put(ByteArray(40) { 1 }) + store.put(ByteArray(40) { 2 }) + + val segsDir = dir.resolve("segments") + val segsBefore = segsDir.listDirectoryEntries("*.seg") + .associate { it.fileName.toString() to Files.size(it) } + assertEquals(2, segsBefore.size, "expected 2 segments before compaction, got $segsBefore") + sealedName = segsBefore.keys.min() + sealedBytes = segsBefore.getValue(sealedName) + sealedSegId = sealedName.removeSuffix(".seg").toLong() + + val report = store.compact(eventStore) + + assertEquals(listOf(sealedSegId), report.compactedSegmentIds, "report: $report") + assertNull(report.newSegmentId, "no new segment should be written, got $report") + assertEquals(sealedBytes, report.bytesReclaimed, "report: $report") + + val segsAfter = segsDir.listDirectoryEntries("*.seg") + .map { it.fileName.toString() }.toSet() + assertFalse(sealedName in segsAfter, "sealed segment file should be deleted, got $segsAfter") + assertEquals(1, segsAfter.size, "only active segment should remain, got $segsAfter") + + assertNull(store.get(dead), "dead artifact must not be retrievable") + } + SqliteArtifactIndex(dir.resolve("index.sqlite")).use { reopened -> + assertTrue( + reopened.entriesIn(sealedSegId).isEmpty(), + "index entries for sealed segment must be removed", + ) + } + } + + @Test + fun `never touches the active segment`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64L * 1024 * 1024).use { store -> + val eventStore = InMemoryEventStore() + store.put("x".toByteArray()) + store.put("y".toByteArray()) + store.put("z".toByteArray()) + + val segsDir = dir.resolve("segments") + val before = segsDir.listDirectoryEntries("*.seg") + .associate { it.fileName.toString() to Files.size(it) } + + val report = store.compact(eventStore) + assertTrue(report.compactedSegmentIds.isEmpty(), "active must not be compacted, got $report") + + val after = segsDir.listDirectoryEntries("*.seg") + .associate { it.fileName.toString() to Files.size(it) } + assertEquals(before, after) + } + } + + @Test + fun `liveness scanner counts both prompt and response ids`(@TempDir dir: Path): Unit = runBlocking { + val eventStore = InMemoryEventStore() + val id1: ArtifactId + val id2: ArtifactId + newStore(dir).use { store -> + id1 = store.put("p1".toByteArray()) + id2 = store.put("p2".toByteArray()) + emitStarted(eventStore, id1) + emitCompleted(eventStore, id2) + } + val index = SqliteArtifactIndex(dir.resolve("index.sqlite")) + index.use { + val scanner = LivenessScanner(eventStore, index) + val live = scanner.scan().liveIds + assertEquals(setOf(id1, id2), live) + } + } +} diff --git a/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/evict/EvictorTest.kt b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/evict/EvictorTest.kt new file mode 100644 index 00000000..55ff5857 --- /dev/null +++ b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/evict/EvictorTest.kt @@ -0,0 +1,267 @@ +package com.correx.infrastructure.artifactscas.evict + +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.NewEvent +import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.inference.TokenUsage +import com.correx.core.utils.TypeId +import com.correx.infrastructure.artifactscas.CasArtifactStore +import com.correx.infrastructure.artifactscas.config.CasConfig +import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex +import com.correx.infrastructure.persistence.InMemoryEventStore +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +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 kotlin.io.path.listDirectoryEntries + +class EvictorTest { + + private val sessionId: SessionId = TypeId("session-evict") + + private fun newStore(dir: Path, maxSegmentBytes: Long, maxTotalBytes: Long): CasArtifactStore { + val index = SqliteArtifactIndex(dir.resolve("index.sqlite")) + return CasArtifactStore(CasConfig(dir, maxSegmentBytes, maxTotalBytes), index) + } + + private suspend fun emitCompleted(store: InMemoryEventStore, response: ArtifactId) { + store.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = InferenceCompletedEvent( + requestId = TypeId(UUID.randomUUID().toString()), + sessionId = sessionId, + stageId = TypeId("stage"), + providerId = TypeId("prov"), + tokensUsed = TokenUsage(0, 0), + latencyMs = 0L, + responseArtifactId = response, + ), + ) + ) + } + + private fun segNames(dir: Path): Set = + dir.resolve("segments").listDirectoryEntries("*.seg").map { it.fileName.toString() }.toSet() + + private fun segIds(dir: Path): List = + dir.resolve("segments").listDirectoryEntries("*.seg") + .mapNotNull { runCatching { it.fileName.toString().removeSuffix(".seg").toLong() }.getOrNull() } + .sorted() + + @Test + fun `under cap - no eviction`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 10_000).use { store -> + val eventStore = InMemoryEventStore() + store.put(ByteArray(40) { 1 }) + store.put(ByteArray(40) { 2 }) + store.put(ByteArray(40) { 3 }) + + val before = segNames(dir) + val report = store.evict(eventStore) + + assertTrue(report.evictedSegmentIds.isEmpty(), "report: $report") + assertEquals(0L, report.bytesReclaimed) + assertEquals(before, segNames(dir)) + } + } + + @Test + fun `over cap, oldest dead segment evicted first`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 80).use { store -> + val eventStore = InMemoryEventStore() + store.put(ByteArray(40) { 1 }) // dead, oldest + store.put(ByteArray(40) { 2 }) // dead + store.put(ByteArray(40) { 3 }) // active segment + + val idsBefore = segIds(dir) + assertTrue(idsBefore.size >= 3, "expected rotation, got $idsBefore") + val oldest = idsBefore.first() + + val report = store.evict(eventStore) + + assertTrue(oldest in report.evictedSegmentIds, "expected oldest=$oldest evicted, got $report") + assertTrue(report.bytesReclaimed > 0, "report: $report") + assertFalse( + "%08d.seg".format(oldest) in segNames(dir), + "oldest segment file should be deleted", + ) + } + SqliteArtifactIndex(dir.resolve("index.sqlite")).use { reopened -> + val oldestId = 1L + assertTrue( + reopened.entriesIn(oldestId).isEmpty(), + "index entries for evicted segment must be removed", + ) + } + } + + @Test + fun `live segment never evicted even if oldest`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 50).use { store -> + val eventStore = InMemoryEventStore() + val live = store.put(ByteArray(40) { 1 }) // oldest, but LIVE + store.put(ByteArray(40) { 2 }) // dead + store.put(ByteArray(40) { 3 }) // active + emitCompleted(eventStore, live) + + val idsBefore = segIds(dir) + val oldest = idsBefore.first() + + val report = store.evict(eventStore) + + assertFalse( + oldest in report.evictedSegmentIds, + "live oldest must not be evicted, got $report", + ) + assertTrue( + "%08d.seg".format(oldest) in segNames(dir), + "live oldest segment file must remain", + ) + } + } + + @Test + fun `active segment never evicted even if dead`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 1).use { store -> + val eventStore = InMemoryEventStore() + store.put(ByteArray(40) { 1 }) + store.put(ByteArray(40) { 2 }) + store.put(ByteArray(40) { 3 }) // active + + val idsBefore = segIds(dir) + val active = idsBefore.last() + + val report = store.evict(eventStore) + + assertFalse( + active in report.evictedSegmentIds, + "active must not be evicted, got $report", + ) + assertTrue( + "%08d.seg".format(active) in segNames(dir), + "active segment file must remain", + ) + } + } + + @Test + fun `evicts strictly oldest-first when multiple dead segments exist`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 200).use { store -> + val eventStore = InMemoryEventStore() + store.put(ByteArray(40) { 1 }) // dead, oldest + store.put(ByteArray(40) { 2 }) // dead + store.put(ByteArray(40) { 3 }) // dead, youngest dead + store.put(ByteArray(40) { 4 }) // active + + val idsBefore = segIds(dir) + assertTrue(idsBefore.size >= 4, "expected >=4 segments, got $idsBefore") + val oldest = idsBefore[0] + val secondOldest = idsBefore[1] + val youngestDead = idsBefore[2] + + val report = store.evict(eventStore) + + assertEquals( + listOf(oldest, secondOldest), + report.evictedSegmentIds, + "expected strictly oldest-first eviction, got $report", + ) + assertTrue( + "%08d.seg".format(youngestDead) in segNames(dir), + "youngest dead segment must survive, ids=${segIds(dir)}", + ) + } + } + + @Test + fun `live segment in the middle is skipped and younger dead segment is evicted`( + @TempDir dir: Path, + ): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 50).use { store -> + val eventStore = InMemoryEventStore() + store.put(ByteArray(40) { 1 }) // dead, oldest + val live = store.put(ByteArray(40) { 2 }) // live, middle + store.put(ByteArray(40) { 3 }) // dead, newer + store.put(ByteArray(40) { 4 }) // active + emitCompleted(eventStore, live) + + val idsBefore = segIds(dir) + assertTrue(idsBefore.size >= 4, "expected >=4 segments, got $idsBefore") + val oldest = idsBefore[0] + val middle = idsBefore[1] + val newerDead = idsBefore[2] + + val report = store.evict(eventStore) + + assertFalse( + middle in report.evictedSegmentIds, + "live middle segment must not be evicted, got $report", + ) + assertTrue( + "%08d.seg".format(middle) in segNames(dir), + "live middle segment file must remain", + ) + assertTrue( + oldest in report.evictedSegmentIds, + "expected oldest dead evicted, got $report", + ) + // If more than one eviction was needed, the next must be the newer dead — in age order. + if (report.evictedSegmentIds.size >= 2) { + assertEquals( + listOf(oldest, newerDead), + report.evictedSegmentIds, + "evictions must be in age order, skipping live middle", + ) + } + } + } + + @Test + fun `evicts multiple segments until under cap`(@TempDir dir: Path): Unit = runBlocking { + newStore(dir, maxSegmentBytes = 64, maxTotalBytes = 100).use { store -> + val eventStore = InMemoryEventStore() + store.put(ByteArray(40) { 1 }) + store.put(ByteArray(40) { 2 }) + store.put(ByteArray(40) { 3 }) + store.put(ByteArray(40) { 4 }) + store.put(ByteArray(40) { 5 }) // active + + val idsBefore = segIds(dir) + assertTrue(idsBefore.size >= 4, "expected several segments, got $idsBefore") + + val report = store.evict(eventStore) + + assertTrue(report.evictedSegmentIds.size >= 2, "expected >=2 evicted, got $report") + + val segsDir = dir.resolve("segments") + val totalAfter = segsDir.listDirectoryEntries("*.seg").sumOf { Files.size(it) } + assertTrue( + totalAfter <= 100 || segIds(dir).size == 1, + "expected under cap or only active remaining, totalAfter=$totalAfter ids=${segIds(dir)}", + ) + for (evicted in report.evictedSegmentIds) { + assertFalse( + "%08d.seg".format(evicted) in segNames(dir), + "evicted segment $evicted must be deleted", + ) + } + } + } +} diff --git a/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScannerTest.kt b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScannerTest.kt new file mode 100644 index 00000000..1fb4352d --- /dev/null +++ b/infrastructure/artifacts-cas/src/test/kotlin/com/correx/infrastructure/artifactscas/recovery/TailScannerTest.kt @@ -0,0 +1,81 @@ +package com.correx.infrastructure.artifactscas.recovery + +import com.correx.infrastructure.artifactscas.CasArtifactStore +import com.correx.infrastructure.artifactscas.config.CasConfig +import com.correx.infrastructure.artifactscas.index.SqliteArtifactIndex +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertArrayEquals +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +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.sql.DriverManager +import kotlin.io.path.listDirectoryEntries + +class TailScannerTest { + + private fun newIndex(dir: Path) = SqliteArtifactIndex(dir.resolve("index.sqlite")) + + @Test + fun `tail beyond known index is re-added on open`(@TempDir dir: Path): Unit = runBlocking { + val payload = "recover me".toByteArray() + val id = CasArtifactStore.open(CasConfig(dir), newIndex(dir)).use { store -> + val id = store.put(payload) + store.flushBefore {} + id + } + // Simulate lost index: wipe the rows. + DriverManager.getConnection("jdbc:sqlite:${dir.resolve("index.sqlite").toAbsolutePath()}").use { conn -> + conn.createStatement().use { it.execute("DELETE FROM artifacts;") } + } + CasArtifactStore.open(CasConfig(dir), newIndex(dir)).use { reopened -> + assertArrayEquals(payload, reopened.get(id)) + } + } + + @Test + fun `corrupt tail is truncated on open`(@TempDir dir: Path): Unit = runBlocking { + val p1 = "first".toByteArray() + val p2 = "second".toByteArray() + val (id1, id2) = CasArtifactStore.open(CasConfig(dir), newIndex(dir)).use { store -> + val a = store.put(p1) + val b = store.put(p2) + store.flushBefore {} + a to b + } + val segFile = dir.resolve("segments").listDirectoryEntries("*.seg").single() + val cleanSize = Files.size(segFile) + // Append garbage that will fail header/length validation. + Files.write( + segFile, + ByteArray(GARBAGE_BYTES) { 0xAB.toByte() }, + java.nio.file.StandardOpenOption.APPEND, + ) + CasArtifactStore.open(CasConfig(dir), newIndex(dir)).use { reopened -> + assertArrayEquals(p1, reopened.get(id1)) + assertArrayEquals(p2, reopened.get(id2)) + } + assertEquals(cleanSize, Files.size(segFile)) + } + + @Test + fun `recover returns null when no segments dir exists`(@TempDir dir: Path): Unit = runBlocking { + val root = dir.resolve("fresh") + Files.createDirectories(root) + newIndex(root).use { idx -> + val scanner = TailScanner(root.resolve("segments"), idx) + assertNull(scanner.recover()) + } + // And open() still succeeds despite no prior segments. + CasArtifactStore.open(CasConfig(root), newIndex(root)).use { store -> + assertNotNull(store) + } + } + + private companion object { + const val GARBAGE_BYTES = 20 + } +} diff --git a/infrastructure/build.gradle b/infrastructure/build.gradle index 90c5e05b..5c5c5342 100644 --- a/infrastructure/build.gradle +++ b/infrastructure/build.gradle @@ -18,6 +18,8 @@ dependencies { implementation project(":infrastructure:tools:filesystem") implementation project(":infrastructure:workflow") implementation project(":core:artifacts") + implementation project(":core:artifacts-store") + implementation project(":infrastructure:artifacts-cas") implementation "io.ktor:ktor-client-core:$ktor_version" implementation "io.ktor:ktor-client-cio:$ktor_version" implementation "io.ktor:ktor-client-content-negotiation:$ktor_version" diff --git a/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt b/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt index 93839e61..453c5f80 100644 --- a/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt +++ b/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/DefaultModelManager.kt @@ -30,12 +30,13 @@ import java.io.File * * Thread-safe: all public methods use Mutex for synchronization. */ -@Suppress("LongParameterList", "TooGenericExceptionCaught", "UnusedPrivateProperty") +@Suppress("LongParameterList", "TooGenericExceptionCaught") class DefaultModelManager( private val llamaServerBin: String = "llama-server", private val host: String = "127.0.0.1", private val port: Int = 10000, private val healthTimeoutMs: Long = 30_000L, + @Suppress("UnusedPrivateProperty") private val eventStore: EventStore, private val httpClient: HttpClient, private val eventDispatcher: EventDispatcher, diff --git a/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/LlamaCppInferenceProvider.kt b/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/LlamaCppInferenceProvider.kt index 78c1d4c5..42b70d20 100644 --- a/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/LlamaCppInferenceProvider.kt +++ b/infrastructure/inference/llama_cpp/src/main/kotlin/com/correx/infrastructure/inference/llama/cpp/LlamaCppInferenceProvider.kt @@ -15,6 +15,7 @@ import com.correx.infrastructure.inference.commons.ModelDescriptor import io.ktor.client.HttpClient import io.ktor.client.call.body import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.HttpTimeout import io.ktor.client.plugins.contentnegotiation.ContentNegotiation import io.ktor.client.request.accept import io.ktor.client.request.get @@ -25,6 +26,8 @@ import io.ktor.http.contentType import io.ktor.serialization.kotlinx.json.json import kotlinx.serialization.json.Json +private const val DEFAULT_REQUEST_TIMEOUT_MS = 60_000L + private fun defaultHttpClient(): HttpClient = HttpClient(CIO) { install(ContentNegotiation) { json( @@ -35,6 +38,9 @@ private fun defaultHttpClient(): HttpClient = HttpClient(CIO) { }, ) } + install(HttpTimeout) { + requestTimeoutMillis = DEFAULT_REQUEST_TIMEOUT_MS + } } @Suppress("TooGenericExceptionCaught", "MagicNumber") diff --git a/infrastructure/persistence/build.gradle b/infrastructure/persistence/build.gradle index 17907e75..a00a2cb5 100644 --- a/infrastructure/persistence/build.gradle +++ b/infrastructure/persistence/build.gradle @@ -9,6 +9,7 @@ dependencies { implementation(project(":core:events")) implementation(project(":core:sessions")) implementation(project(":core:artifacts")) + implementation(project(":core:artifacts-store")) implementation "org.xerial:sqlite-jdbc" testImplementation(testFixtures(project(":testing:contracts"))) testImplementation "org.junit.jupiter:junit-jupiter" diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt index c8240c84..7357af04 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt @@ -16,7 +16,7 @@ class InMemoryEventStore : EventStore { private val seenEventIds = ConcurrentHashMap.newKeySet() private val subscriptions = ConcurrentHashMap>() - override fun append(event: NewEvent): StoredEvent { + override suspend fun append(event: NewEvent): StoredEvent { val stream = streams.computeIfAbsent(event.metadata.sessionId) { mutableListOf() } synchronized(stream) { @@ -27,7 +27,7 @@ class InMemoryEventStore : EventStore { } } - override fun appendAll(events: List): List { + override suspend fun appendAll(events: List): List { if (events.isEmpty()) return emptyList() val sessionId = events.first().metadata.sessionId @@ -58,6 +58,11 @@ class InMemoryEventStore : EventStore { override fun subscribe(sessionId: SessionId): Flow = subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) } + override fun allEvents(): Sequence = + streams.values + .flatMap { stream -> synchronized(stream) { stream.toList() } } + .asSequence() + private fun doAppend(event: NewEvent, stream: MutableList): StoredEvent { val seq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) } .incrementAndGet() diff --git a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt index 63373936..f83eec56 100644 --- a/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt +++ b/infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt @@ -1,5 +1,6 @@ package com.correx.infrastructure.persistence +import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.NewEvent import com.correx.core.events.events.StoredEvent @@ -11,8 +12,10 @@ import com.correx.core.events.types.CorrelationId import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.infrastructure.persistence.util.JDBCHelper.transaction +import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.withContext import kotlinx.datetime.Instant import java.sql.Connection import java.sql.ResultSet @@ -20,7 +23,8 @@ import java.util.concurrent.ConcurrentHashMap class SqliteEventStore( private val connection: Connection, - private val jsonSerializer: JsonEventSerializer = JsonEventSerializer(eventJson) + private val jsonSerializer: JsonEventSerializer = JsonEventSerializer(eventJson), + private val artifactStore: ArtifactStore, ) : EventStore { private val subscriptions = ConcurrentHashMap>() @@ -43,46 +47,57 @@ class SqliteEventStore( } } - override fun append(event: NewEvent): StoredEvent { - val stored = connection.transaction { - val existing = findByEventId(event.metadata.eventId) - check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } + override suspend fun append(event: NewEvent): StoredEvent { + var stored: StoredEvent? = null + artifactStore.flushBefore { + withContext(Dispatchers.IO) { + stored = connection.transaction { + val existing = findByEventId(event.metadata.eventId) + check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } - val seq = nextSequence(event.metadata.sessionId) + val seq = nextSequence(event.metadata.sessionId) - val s = StoredEvent( - metadata = event.metadata, - sequence = seq, - payload = event.payload - ) + val s = StoredEvent( + metadata = event.metadata, + sequence = seq, + payload = event.payload + ) - insert(s) - s + insert(s) + s + } + } } - subscriptions[event.metadata.sessionId]?.tryEmit(stored) - return stored + val result = checkNotNull(stored) + subscriptions[event.metadata.sessionId]?.tryEmit(result) + return result } - override fun appendAll(events: List): List { + override suspend fun appendAll(events: List): List { if (events.isEmpty()) return emptyList() val sessionId = events.first().metadata.sessionId - val stored = connection.transaction { - events.map { event -> - val existing = findByEventId(event.metadata.eventId) - check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } + var stored: List = emptyList() + artifactStore.flushBefore { + withContext(Dispatchers.IO) { + stored = connection.transaction { + events.map { event -> + val existing = findByEventId(event.metadata.eventId) + check(existing == null) { "duplicate event_id: ${event.metadata.eventId}" } - val seq = nextSequence(sessionId) + val seq = nextSequence(sessionId) - val s = StoredEvent( - metadata = event.metadata, - sequence = seq, - payload = event.payload - ) + val s = StoredEvent( + metadata = event.metadata, + sequence = seq, + payload = event.payload + ) - insert(s) - s + insert(s) + s + } + } } } val flow = subscriptions[sessionId] @@ -103,7 +118,7 @@ class SqliteEventStore( ps.executeQuery().use { rs -> buildList { while (rs.next()) { - add(map(rs)) + add(rs.toStoredEvent(jsonSerializer)) } } } @@ -124,7 +139,7 @@ class SqliteEventStore( ps.executeQuery().use { rs -> buildList { while (rs.next()) { - add(map(rs)) + add(rs.toStoredEvent(jsonSerializer)) } } } @@ -149,6 +164,22 @@ class SqliteEventStore( override fun subscribe(sessionId: SessionId): Flow = subscriptions.computeIfAbsent(sessionId) { MutableSharedFlow(replay = 0, extraBufferCapacity = 64) } + override fun allEvents(): Sequence = + connection.prepareStatement( + """ + SELECT * FROM events + ORDER BY session_id ASC, sequence ASC + """.trimIndent() + ).use { ps -> + ps.executeQuery().use { rs -> + buildList { + while (rs.next()) { + add(rs.toStoredEvent(jsonSerializer)) + } + } + } + }.asSequence() + // ---------- helpers ---------- private fun nextSequence(sessionId: SessionId): Long = @@ -205,21 +236,22 @@ class SqliteEventStore( ps.setString(1, eventId.value) ps.executeQuery().use { rs -> - if (rs.next()) map(rs) else null + if (rs.next()) rs.toStoredEvent(jsonSerializer) else null } } - private fun map(rs: ResultSet): StoredEvent = - StoredEvent( - metadata = EventMetadata( - eventId = EventId(rs.getString("event_id")), - sessionId = SessionId(rs.getString("session_id")), - timestamp = Instant.parse(rs.getString("timestamp")), - schemaVersion = rs.getInt("schema_version"), - causationId = rs.getString("causation_id")?.let { CausationId(it) }, - correlationId = rs.getString("correlation_id")?.let { CorrelationId(it) } - ), - sequence = rs.getLong("sequence"), - payload = jsonSerializer.deserialize(rs.getString("payload")) - ) -} \ No newline at end of file +} + +private fun ResultSet.toStoredEvent(jsonSerializer: JsonEventSerializer): StoredEvent = + StoredEvent( + metadata = EventMetadata( + eventId = EventId(getString("event_id")), + sessionId = SessionId(getString("session_id")), + timestamp = Instant.parse(getString("timestamp")), + schemaVersion = getInt("schema_version"), + causationId = getString("causation_id")?.let { CausationId(it) }, + correlationId = getString("correlation_id")?.let { CorrelationId(it) } + ), + sequence = getLong("sequence"), + payload = jsonSerializer.deserialize(getString("payload")) + ) \ No newline at end of file diff --git a/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteEventStoreTest.kt b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteEventStoreTest.kt index 3058b3ee..8106c2e6 100644 --- a/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteEventStoreTest.kt +++ b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteEventStoreTest.kt @@ -1,12 +1,13 @@ package com.correx.infrastructure.persistence import com.correx.core.events.stores.EventStore +import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore import com.correx.testing.contracts.fixtures.events.store.EventStoreContractTest import java.sql.DriverManager class SqliteEventStoreTest : EventStoreContractTest() { override fun store(): EventStore { val conn = DriverManager.getConnection("jdbc:sqlite::memory:") - return SqliteEventStore(conn) + return SqliteEventStore(conn, artifactStore = NoopArtifactStore()) } } \ No newline at end of file diff --git a/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteReplayTest.kt b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteReplayTest.kt index f8a02b2e..5ac7d2fc 100644 --- a/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteReplayTest.kt +++ b/infrastructure/persistence/src/test/kotlin/com/correx/infrastructure/persistence/SqliteReplayTest.kt @@ -5,6 +5,7 @@ import com.correx.core.sessions.SessionCounterProjection import com.correx.core.sessions.SessionCounterState import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.core.sessions.projections.replay.EventReplayer +import com.correx.testing.contracts.fixtures.artifactstore.NoopArtifactStore import com.correx.testing.contracts.fixtures.projections.SessionReplayContractTest import java.sql.DriverManager @@ -14,7 +15,7 @@ class SqliteReplayTest : SessionReplayContractTest() { val connection = DriverManager.getConnection("jdbc:sqlite::memory:") - return SqliteEventStore(connection) + return SqliteEventStore(connection, artifactStore = NoopArtifactStore()) } override fun replayer( diff --git a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt index a43f4ef4..6ede036e 100644 --- a/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt +++ b/infrastructure/src/main/kotlin/com/correx/infrastructure/InfrastructureModule.kt @@ -1,6 +1,7 @@ package com.correx.infrastructure import com.correx.core.approvals.domain.ApprovalEngine +import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.EventDispatcher import com.correx.core.events.stores.EventStore import com.correx.core.inference.DefaultInferenceRouter @@ -17,6 +18,10 @@ 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.tools.DefaultToolRegistry import com.correx.infrastructure.tools.DispatchingToolExecutor @@ -31,17 +36,32 @@ import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository import com.correx.core.artifacts.repository.ArtifactRepository import com.correx.core.artifacts.DefaultArtifactReducer import io.ktor.client.HttpClient +import java.nio.file.Files import java.nio.file.Path +import java.nio.file.Paths import java.sql.DriverManager +@Suppress("TooManyFunctions") object InfrastructureModule { private val defaultDbPath: String = "${System.getProperty("user.home")}/.config/correx/correx.db" - fun createEventStore(dbPath: String = defaultDbPath): EventStore { + private val defaultArtifactsRoot: String = + "${System.getProperty("user.home")}/.config/correx/artifacts" + + fun createEventStore(artifactStore: ArtifactStore, dbPath: String = defaultDbPath): EventStore { val dir = java.io.File(dbPath).parentFile if (!dir.exists()) dir.mkdirs() - return SqliteEventStore(DriverManager.getConnection("jdbc:sqlite:$dbPath")) + return SqliteEventStore( + connection = DriverManager.getConnection("jdbc:sqlite:$dbPath"), + artifactStore = artifactStore, + ) + } + + fun createArtifactStore(rootDir: Path = Paths.get(defaultArtifactsRoot)): ArtifactStore { + Files.createDirectories(rootDir) + val index = SqliteArtifactIndex(rootDir.resolve("index.sqlite")) + return runBlocking { CasArtifactStore.open(CasConfig(rootDir), index) } } fun createModelManager( diff --git a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt index 41b8ef86..08fc6f79 100644 --- a/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt +++ b/infrastructure/workflow/src/main/kotlin/com/correx/infrastructure/workflow/TomlWorkflowLoader.kt @@ -24,6 +24,7 @@ private data class WorkflowFile( private data class StageSection( val id: String = "", val prompt: String? = null, + val systemPrompt: String? = null, val produces: List = emptyList(), val needs: List = emptyList(), @JsonProperty("allowed_tools") val allowedTools: List = emptyList(), @@ -66,7 +67,10 @@ class TomlWorkflowLoader : WorkflowLoader { allowedTools = s.allowedTools.toSet(), tokenBudget = s.tokenBudget, maxRetries = s.maxRetries, - metadata = buildMap { s.prompt?.let { put("prompt", it) } }, + metadata = buildMap { + s.prompt?.let { put("prompt", it) } + ?: s.systemPrompt?.let { put("systemPrompt", it) } + }, ) } diff --git a/settings.gradle b/settings.gradle index 13e418c8..10a64d87 100644 --- a/settings.gradle +++ b/settings.gradle @@ -21,6 +21,7 @@ include ':core:inference' include ':core:stages' include ':core:agents' include ':core:artifacts' +include ':core:artifacts-store' include ':core:validation' include ':core:approvals' include ':core:tools' @@ -41,6 +42,7 @@ include ':infrastructure:security' include ':infrastructure:scheduler' include ':infrastructure:telemetry' include ':infrastructure:workflow' +include ':infrastructure:artifacts-cas' include ':interfaces:api' include ':interfaces:cli' diff --git a/testing/contracts/build.gradle b/testing/contracts/build.gradle index 484dacea..79447679 100644 --- a/testing/contracts/build.gradle +++ b/testing/contracts/build.gradle @@ -17,7 +17,9 @@ dependencies { testImplementation(project(":core:context")) testFixturesImplementation(project(":core:events")) testFixturesImplementation(project(":core:sessions")) + testFixturesImplementation(project(":core:artifacts-store")) testFixturesImplementation(project(":testing:fixtures")) testFixturesImplementation "org.jetbrains.kotlinx:kotlinx-datetime:$kotlinx_datetime_version" testFixturesImplementation "org.junit.jupiter:junit-jupiter:$junit_version" + testFixturesImplementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinx_coroutines_version" } diff --git a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/artifactstore/NoopArtifactStore.kt b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/artifactstore/NoopArtifactStore.kt new file mode 100644 index 00000000..0281dc79 --- /dev/null +++ b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/artifactstore/NoopArtifactStore.kt @@ -0,0 +1,11 @@ +package com.correx.testing.contracts.fixtures.artifactstore + +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.types.ArtifactId +import com.correx.core.utils.TypeId + +class NoopArtifactStore : ArtifactStore { + override suspend fun put(bytes: ByteArray): ArtifactId = TypeId("00".repeat(32)) + override suspend fun get(id: ArtifactId): ByteArray? = null + override suspend fun flushBefore(commit: suspend () -> Unit) { commit() } +} diff --git a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/events/store/EventStoreContractTest.kt b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/events/store/EventStoreContractTest.kt index ad4c035e..49687588 100644 --- a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/events/store/EventStoreContractTest.kt +++ b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/events/store/EventStoreContractTest.kt @@ -5,6 +5,7 @@ import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.testing.contracts.utils.ConcurrencyRunner import com.correx.testing.fixtures.EventFixtures +import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions import org.junit.jupiter.api.Test import org.junit.jupiter.api.assertThrows @@ -13,7 +14,7 @@ abstract class EventStoreContractTest { protected abstract fun store(): EventStore @Test - fun `appends preserve ordering`() { + fun `appends preserve ordering`(): Unit = runBlocking { val store = store() store.append(EventFixtures.newEvent(EventId("e1"), SessionId("s1"))) @@ -27,19 +28,19 @@ abstract class EventStoreContractTest { } @Test - fun `idempotency prevents duplicates`() { + fun `idempotency prevents duplicates`(): Unit = runBlocking { val store = store() val e = EventFixtures.newEvent(EventId("dup"), SessionId("s1")) store.append(e) assertThrows { - store.append(e) + runBlocking { store.append(e) } } Assertions.assertEquals(1, store.read(SessionId("s1")).size) } @Test - fun `sessions are isolated`() { + fun `sessions are isolated`(): Unit = runBlocking { val store = store() store.append(EventFixtures.newEvent(EventId("a"), SessionId("s1"))) @@ -50,7 +51,7 @@ abstract class EventStoreContractTest { } @Test - fun `readFrom respects sequence cursor`() { + fun `readFrom respects sequence cursor`(): Unit = runBlocking { val store = store() store.append(EventFixtures.newEvent(EventId("1"), SessionId("s1"))) @@ -67,9 +68,11 @@ abstract class EventStoreContractTest { val store = store() ConcurrencyRunner.run(20, 100) { t, i -> - store.append( - EventFixtures.newEvent(EventId("e-$i-$t"), SessionId("s1")) - ) + runBlocking { + store.append( + EventFixtures.newEvent(EventId("e-$i-$t"), SessionId("s1")) + ) + } } val all = store.read(SessionId("s1")) @@ -80,7 +83,7 @@ abstract class EventStoreContractTest { } @Test - fun `sequences are strictly increasing per session`() { + fun `sequences are strictly increasing per session`(): Unit = runBlocking { val store = store() repeat(100) { @@ -95,7 +98,7 @@ abstract class EventStoreContractTest { } @Test - fun `sequences are isolated per session`() { + fun `sequences are isolated per session`(): Unit = runBlocking { val store = store() repeat(50) { @@ -111,7 +114,7 @@ abstract class EventStoreContractTest { } @Test - fun `appendAll is atomic per session`() { + fun `appendAll is atomic per session`(): Unit = runBlocking { val store = store() val batch = (1..50).map { @@ -133,12 +136,14 @@ abstract class EventStoreContractTest { val store = store() ConcurrencyRunner.run(10, 50) { t, i -> - if (i % 2 == 0) { - store.append(EventFixtures.newEvent(EventId("a-$t-$i"), SessionId("s1"))) - } else { - store.appendAll( - listOf(EventFixtures.newEvent(EventId("b-$t-$i"), SessionId("s1"))) - ) + runBlocking { + if (i % 2 == 0) { + store.append(EventFixtures.newEvent(EventId("a-$t-$i"), SessionId("s1"))) + } else { + store.appendAll( + listOf(EventFixtures.newEvent(EventId("b-$t-$i"), SessionId("s1"))) + ) + } } } @@ -150,7 +155,7 @@ abstract class EventStoreContractTest { } @Test - fun `read returns stable snapshot`() { + fun `read returns stable snapshot`(): Unit = runBlocking { val store = store() repeat(100) { @@ -164,7 +169,7 @@ abstract class EventStoreContractTest { } @Test - fun `readFrom behaves like streaming cursor`() { + fun `readFrom behaves like streaming cursor`(): Unit = runBlocking { val store = store() repeat(10) { @@ -177,4 +182,4 @@ abstract class EventStoreContractTest { Assertions.assertTrue(first.all { it.sequence > 3 }) Assertions.assertTrue(second.all { it.sequence > 7 }) } -} \ No newline at end of file +} diff --git a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/ReplayContractTest.kt b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/ReplayContractTest.kt index 9cc6a615..c7c98cc0 100644 --- a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/ReplayContractTest.kt +++ b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/ReplayContractTest.kt @@ -5,6 +5,7 @@ import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.sessions.projections.replay.EventReplayer import com.correx.testing.fixtures.EventFixtures.newEvent +import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test @@ -17,7 +18,7 @@ abstract class ReplayContractTest { ): EventReplayer @Test - fun `rebuild is deterministic`() { + fun `rebuild is deterministic`(): Unit = runBlocking { val store = store() repeat(50) { diff --git a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/SessionReplayContractTest.kt b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/SessionReplayContractTest.kt index 9d0c45d0..ea378bd5 100644 --- a/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/SessionReplayContractTest.kt +++ b/testing/contracts/src/testFixtures/kotlin/com/correx/testing/contracts/fixtures/projections/SessionReplayContractTest.kt @@ -6,6 +6,7 @@ import com.correx.core.events.types.SessionId import com.correx.core.sessions.SessionCounterState import com.correx.core.sessions.projections.replay.EventReplayer import com.correx.testing.fixtures.EventFixtures.newEvent +import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test @@ -18,7 +19,7 @@ abstract class SessionReplayContractTest { ): EventReplayer @Test - fun `rebuild is deterministic`() { + fun `rebuild is deterministic`(): Unit = runBlocking { val store = store() repeat(50) { @@ -34,7 +35,7 @@ abstract class SessionReplayContractTest { } @Test - fun `rebuild reflects complete stream`() { + fun `rebuild reflects complete stream`(): Unit = runBlocking { val store = store() repeat(42) { diff --git a/testing/deterministic/src/test/kotlin/SessionReplayDeterminismTest.kt b/testing/deterministic/src/test/kotlin/SessionReplayDeterminismTest.kt index a72477d5..64eba35d 100644 --- a/testing/deterministic/src/test/kotlin/SessionReplayDeterminismTest.kt +++ b/testing/deterministic/src/test/kotlin/SessionReplayDeterminismTest.kt @@ -11,6 +11,7 @@ import com.correx.core.sessions.DefaultSessionReducer import com.correx.core.sessions.SessionProjector import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.infrastructure.persistence.InMemoryEventStore +import kotlinx.coroutines.runBlocking import kotlinx.datetime.Clock import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test @@ -23,7 +24,7 @@ class SessionReplayDeterminismTest { ) @Test - fun `same events produce same state`() { + fun `same events produce same state`(): Unit = runBlocking { val sessionId = SessionId("s1") val store1 = InMemoryEventStore() diff --git a/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt b/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt index f06d8c50..1ac43c7b 100644 --- a/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt +++ b/testing/fixtures/src/main/kotlin/com/correx/testing/fixtures/InferenceFixtures.kt @@ -4,12 +4,14 @@ import com.correx.core.context.model.ContextPack import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.NewEvent +import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.ContextPackId import com.correx.core.events.types.EventId import com.correx.core.events.types.InferenceRequestId import com.correx.core.events.types.ProviderId import com.correx.core.events.types.SessionId 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 @@ -73,6 +75,9 @@ object InferenceFixtures { ) } + private const val PLACEHOLDER_HEX_BYTES = 32 + private val PLACEHOLDER_ARTIFACT_ID: ArtifactId = TypeId("00".repeat(PLACEHOLDER_HEX_BYTES)) + fun inferenceCompleted( requestId: InferenceRequestId, sessionId: SessionId, @@ -80,6 +85,7 @@ object InferenceFixtures { providerId: ProviderId, tokensUsed: TokenUsage, latencyMs: Long, + responseArtifactId: ArtifactId = PLACEHOLDER_ARTIFACT_ID, ): InferenceCompletedEvent { return InferenceCompletedEvent( requestId = requestId, @@ -88,6 +94,7 @@ object InferenceFixtures { providerId = providerId, tokensUsed = tokensUsed, latencyMs = latencyMs, + responseArtifactId = responseArtifactId, ) } @@ -127,6 +134,7 @@ object InferenceFixtures { providerId: ProviderId, tokensUsed: TokenUsage, latencyMs: Long, + responseArtifactId: ArtifactId = PLACEHOLDER_ARTIFACT_ID, ): NewEvent { return NewEvent( metadata = EventMetadata( @@ -144,6 +152,7 @@ object InferenceFixtures { providerId = providerId, tokensUsed = tokensUsed, latencyMs = latencyMs, + responseArtifactId = responseArtifactId, ), ) } diff --git a/testing/integration/build.gradle b/testing/integration/build.gradle index 7416239a..bffaf03e 100644 --- a/testing/integration/build.gradle +++ b/testing/integration/build.gradle @@ -16,8 +16,10 @@ dependencies { testImplementation(project(":core:risk")) testImplementation(project(":infrastructure:persistence")) testImplementation(project(":core:artifacts")) + testImplementation(project(":core:artifacts-store")) testImplementation(project(":testing:fixtures")) testImplementation(project(":testing:kernel")) testImplementation(project(":testing:contracts")) + testImplementation(testFixtures(project(":testing:contracts"))) testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.9.0") } \ No newline at end of file diff --git a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt index 186b5a9f..54903224 100644 --- a/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt +++ b/testing/integration/src/test/kotlin/SessionOrchestratorIntegrationTest.kt @@ -43,6 +43,12 @@ 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 @@ -78,6 +84,7 @@ class SessionOrchestratorIntegrationTest { }, ) private val riskAssessor = DefaultRiskAssessor() + private val artifactStore = NoopArtifactStore() private val repositories = OrchestratorRepositories( eventStore = eventStore, @@ -100,6 +107,7 @@ class SessionOrchestratorIntegrationTest { repositories = repositories, engines = engines, retryCoordinator = retryCoordinator, + artifactStore = artifactStore, ) @Test @@ -112,7 +120,6 @@ class SessionOrchestratorIntegrationTest { orchestrator.run(sessionId, graph, config) val events = eventStore.read(sessionId) - assertEquals(2, events.size) val workflowStarted = events.find { it.payload is WorkflowStartedEvent } assertNotNull(workflowStarted) @@ -147,6 +154,7 @@ class SessionOrchestratorIntegrationTest { }, ), retryCoordinator = retryCoordinator, + artifactStore = artifactStore, ) failingOrchestrator.run(sessionId, graph, config) @@ -173,6 +181,7 @@ class SessionOrchestratorIntegrationTest { repositories = repositories, engines = engines.copy(validationPipeline = approvingPipeline), retryCoordinator = retryCoordinator, + artifactStore = artifactStore, ) val runJob = launch { orchestrator.run(sessionId, graph, config) } @@ -197,7 +206,20 @@ class SessionOrchestratorIntegrationTest { fun `workflow resumes after approval`(): Unit = runBlocking { val sessionId = SessionId("s4") val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 3, backoffMs = 0)) - val graph = threeStageGraph() + // 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( + stages = mapOf(StageId("A") to StageConfig()), + transitions = setOf( + com.correx.core.transitions.graph.TransitionEdge( + id = com.correx.core.events.types.TransitionId("t1"), + from = StageId("A"), + to = StageId("done"), + condition = { true }, + ), + ), + start = StageId("A"), + ) val approvingPipeline = ValidationPipeline( validators = listOf(cyclePolicyMissingValidator()), approvalTrigger = ApprovalTrigger(), @@ -206,6 +228,7 @@ class SessionOrchestratorIntegrationTest { repositories = repositories, engines = engines.copy(validationPipeline = approvingPipeline), retryCoordinator = retryCoordinator, + artifactStore = artifactStore, ) val runJob = launch { approvalOrchestrator.run(sessionId, graph, config) } @@ -257,6 +280,7 @@ class SessionOrchestratorIntegrationTest { repositories = repositories, engines = engines, retryCoordinator = retryCoordinator, + artifactStore = artifactStore, ) val sessionId = SessionId("s6") @@ -269,6 +293,30 @@ class SessionOrchestratorIntegrationTest { assertNotNull(failed.reason) } + @Test + fun `artifactStore put is called for prompt and response and ids appear on inference events`(): Unit = runBlocking { + val recordingStore = RecordingArtifactStore() + val sessionId = SessionId("s-artifact") + val config = OrchestrationConfig(retryPolicy = RetryPolicy(maxAttempts = 1, backoffMs = 0)) + val recordingOrchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = retryCoordinator, + artifactStore = recordingStore, + ) + + recordingOrchestrator.run(sessionId, graph, config) + + 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() + + assertEquals(recordingStore.idForIndex(0), started.promptArtifactId) + assertEquals(recordingStore.idForIndex(1), completed.responseArtifactId) + } + @Test fun `run throws IllegalArgumentException when transition targets undeclared stage`(): Unit = runBlocking { val a = StageId("A") @@ -327,3 +375,21 @@ class SessionOrchestratorIntegrationTest { } } + +private class RecordingArtifactStore : ArtifactStore { + val puts: MutableList = mutableListOf() + private val byId: MutableMap = mutableMapOf() + + fun idForIndex(i: Int): ArtifactId = TypeId("0".repeat(63) + i.toString(16)) + + override suspend fun put(bytes: ByteArray): ArtifactId { + val id = idForIndex(puts.size) + puts.add(bytes) + byId[id] = bytes + return id + } + + override suspend fun get(id: ArtifactId): ByteArray? = byId[id] + + override suspend fun flushBefore(commit: suspend () -> Unit) { commit() } +} diff --git a/testing/projections/src/test/kotlin/InferenceProjectorTest.kt b/testing/projections/src/test/kotlin/InferenceProjectorTest.kt index 45f28c86..17567996 100644 --- a/testing/projections/src/test/kotlin/InferenceProjectorTest.kt +++ b/testing/projections/src/test/kotlin/InferenceProjectorTest.kt @@ -9,6 +9,7 @@ import com.correx.core.inference.InferenceRecord import com.correx.core.inference.InferenceState import com.correx.core.inference.InferenceStatus import com.correx.core.inference.TokenUsage +import com.correx.core.utils.TypeId import com.correx.testing.fixtures.EventFixtures.stored import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.BeforeEach @@ -38,6 +39,7 @@ class InferenceProjectorTest { sessionId = SessionId("sess1"), stageId = StageId("stageA"), providerId = ProviderId("providerX"), + promptArtifactId = TypeId("00".repeat(32)), ), ) val newState = projector.apply(initialState, event) @@ -67,6 +69,7 @@ class InferenceProjectorTest { providerId = ProviderId("providerX"), tokensUsed = TokenUsage(50, 50), latencyMs = 500L, + responseArtifactId = TypeId("00".repeat(32)), ), ) val newState = projector.apply(initialState, event) @@ -88,6 +91,7 @@ class InferenceProjectorTest { sessionId = SessionId("sess1"), stageId = StageId("stageA"), providerId = ProviderId("providerX"), + promptArtifactId = TypeId("00".repeat(32)), ), ) var currentState = projector.apply(initialState, startEvent) @@ -102,6 +106,7 @@ class InferenceProjectorTest { providerId = ProviderId("providerX"), tokensUsed = TokenUsage(100, 100), latencyMs = 1000L, + responseArtifactId = TypeId("00".repeat(32)), ), ) currentState = projector.apply(currentState, completeEvent) diff --git a/testing/projections/src/test/kotlin/InferenceReducerTest.kt b/testing/projections/src/test/kotlin/InferenceReducerTest.kt index 687723ba..22fc3508 100644 --- a/testing/projections/src/test/kotlin/InferenceReducerTest.kt +++ b/testing/projections/src/test/kotlin/InferenceReducerTest.kt @@ -13,6 +13,7 @@ import com.correx.core.inference.InferenceReducer import com.correx.core.inference.InferenceState import com.correx.core.inference.InferenceStatus import com.correx.core.inference.TokenUsage +import com.correx.core.utils.TypeId import com.correx.testing.fixtures.EventFixtures.stored import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.BeforeEach @@ -36,6 +37,7 @@ class InferenceReducerTest { sessionId = SessionId("sess1"), stageId = StageId("stageA"), providerId = ProviderId("providerX"), + promptArtifactId = TypeId("00".repeat(32)), ), ) val newState = reducer.reduce(initialState, event) @@ -65,6 +67,7 @@ class InferenceReducerTest { providerId = ProviderId("providerX"), tokensUsed = TokenUsage(50, 50), latencyMs = 500L, + responseArtifactId = TypeId("00".repeat(32)), ), ) val newState = reducer.reduce(initialState, event) @@ -176,6 +179,7 @@ class InferenceReducerTest { providerId = ProviderId("providerX"), tokensUsed = TokenUsage(50, 50), latencyMs = 500L, + responseArtifactId = TypeId("00".repeat(32)), ), ) val newState = reducer.reduce(initialState, event) diff --git a/testing/replay/src/test/kotlin/SessionReplayTest.kt b/testing/replay/src/test/kotlin/SessionReplayTest.kt index 68f4e9bc..54667f0d 100644 --- a/testing/replay/src/test/kotlin/SessionReplayTest.kt +++ b/testing/replay/src/test/kotlin/SessionReplayTest.kt @@ -23,7 +23,7 @@ class SessionReplayTest { private val replayer = DefaultEventReplayer(store, projector) @Test - fun `rebuild session from event stream`() { + fun `rebuild session from event stream`(): Unit = kotlinx.coroutines.runBlocking { val sessionId = SessionId("s1") val metadataToPayload = mapOf( diff --git a/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt b/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt index e1345378..66f0b903 100644 --- a/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt +++ b/testing/replay/src/test/kotlin/TransitionReplayIntegrationTest.kt @@ -14,6 +14,7 @@ import com.correx.core.sessions.SessionStatus import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.infrastructure.persistence.InMemoryEventStore import com.correx.testing.fixtures.EventFixtures.newEvent +import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test @@ -22,7 +23,7 @@ class TransitionReplayIntegrationTest { private val sessionId = SessionId("session-1") @Test - fun `workflow replay reconstructs COMPLETED session`() { + fun `workflow replay reconstructs COMPLETED session`(): Unit = runBlocking { val store = InMemoryEventStore() @@ -85,7 +86,7 @@ class TransitionReplayIntegrationTest { } @Test - fun `workflow replay reconstructs FAILED session`() { + fun `workflow replay reconstructs FAILED session`(): Unit = runBlocking { val store = InMemoryEventStore() @@ -144,7 +145,7 @@ class TransitionReplayIntegrationTest { } @Test - fun `replay is deterministic for identical event stream`() { + fun `replay is deterministic for identical event stream`(): Unit = runBlocking { val events = listOf( newEvent( sessionId = sessionId,