feat(cas): content-addressed artifact store (steps 1–8)

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.
2026-05-18 12:22:38 +04:00
parent bbff73108e
commit 219e2c762e
60 changed files with 2042 additions and 165 deletions
@@ -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"
 }
@@ -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<SessionId, AtomicBoolean> =
        ConcurrentHashMap<SessionId, AtomicBoolean>()

@@ -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
                }
            }
        }
@@ -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<SessionId, AtomicBoolean>,
    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)
    }
@@ -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<SessionId, AtomicBoolean>
    internal val pendingApprovals: ConcurrentHashMap<ApprovalRequestId, CompletableDeferred<ApprovalDecision>> =
@@ -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,