diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index 82413d10..a0b47f77 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -188,6 +188,25 @@ class ServerModule( } } + /** + * Rehydrates [artifactContentCache] from durable events and then resumes the session. + * Called by [POST /sessions/{id}/resume] to restart a session after a server restart + * without re-running already-completed stages. + */ + fun launchSessionResumeWithRehydrate(sessionId: SessionId, graph: com.correx.core.transitions.graph.WorkflowGraph) { + activeSessionJobs.computeIfAbsent(sessionId) { + moduleScope.launch { + runCatching { + orchestrator.rehydrate(sessionId) + orchestrator.resume(sessionId, graph, defaultOrchestrationConfig) + }.onFailure { e -> + log.error("resumeSession (rehydrate): session={} failed", sessionId.value, e) + } + activeSessionJobs.remove(sessionId) + } + } + } + private fun resumeAbandonedSessions() { eventStore.allSessionIds().forEach { sessionId -> val orchState = orchestrationRepository.getState(sessionId) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt index 43fe1585..f99ca519 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt @@ -43,6 +43,7 @@ fun Route.sessionRoutes(module: ServerModule) { getSessionRoute(module) cancelSessionRoute(module) undoSessionRoute(module) + resumeSessionRoute(module) getEventsRoute(module) webSocket("/stream") { val id = call.parameters["id"] ?: return@webSocket @@ -98,6 +99,21 @@ private fun Route.undoSessionRoute(module: ServerModule) { } } +private fun Route.resumeSessionRoute(module: ServerModule) { + post("/resume") { + val id = call.parameters["id"] + ?: return@post call.respond(HttpStatusCode.BadRequest, "Missing session id") + val orchState = runCatching { module.orchestrationRepository.getState(TypeId(id)) }.getOrNull() + ?: return@post call.respond(HttpStatusCode.NotFound, "Session not found") + val workflowId = orchState.workflowId.takeIf { it.isNotBlank() } + ?: return@post call.respond(HttpStatusCode.BadRequest, "Session has no workflowId") + val graph = module.workflowRegistry.find(workflowId) + ?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: $workflowId") + module.launchSessionResumeWithRehydrate(TypeId(id), graph) + call.respond(HttpStatusCode.Accepted) + } +} + private fun Route.getEventsRoute(module: ServerModule) { get("/events") { val id = call.parameters["id"] 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 ba095f54..d305ecbf 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 @@ -162,6 +162,22 @@ class DefaultSessionOrchestrator( fun validatedArtifactContent(sessionId: SessionId, artifactId: ArtifactId): String? = artifactContentCache["${sessionId.value}:${artifactId.value}"] + /** + * Rebuilds [artifactContentCache] from durable events after a server restart. + * Scans all events for [ArtifactValidatedEvent] and re-fetches the stored bytes + * from [artifactStore] for each one. Must be called before [resume] to ensure + * transition conditions that read artifact content work correctly. + */ + suspend fun rehydrate(sessionId: SessionId) { + repositories.eventStore.readFrom(sessionId, fromSequence = 0L).forEach { event -> + val p = event.payload as? ArtifactValidatedEvent ?: return@forEach + val key = "${sessionId.value}:${p.artifactId.value}" + artifactStore.get(p.artifactId) + ?.toString(Charsets.UTF_8) + ?.let { artifactContentCache[key] = it } + } + } + suspend fun resume( sessionId: SessionId, graph: WorkflowGraph, 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 96399085..14703f5d 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 @@ -134,7 +134,7 @@ private const val REPO_MAP_INJECT_TOP_K = 30 abstract class SessionOrchestrator( repositories: OrchestratorRepositories, engines: OrchestratorEngines, - private val artifactStore: ArtifactStore, + protected val artifactStore: ArtifactStore, private val decisionJournalRepository: DefaultDecisionJournalRepository, private val decisionJournalRenderer: DecisionJournalRenderer = DecisionJournalRenderer(), ) { diff --git a/testing/integration/src/test/kotlin/RehydrateTest.kt b/testing/integration/src/test/kotlin/RehydrateTest.kt new file mode 100644 index 00000000..22b1edef --- /dev/null +++ b/testing/integration/src/test/kotlin/RehydrateTest.kt @@ -0,0 +1,140 @@ +import com.correx.core.approvals.ApprovalProjector +import com.correx.core.approvals.DefaultApprovalReducer +import com.correx.core.approvals.DefaultApprovalRepository +import com.correx.core.approvals.domain.DefaultApprovalEngine +import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.artifacts.DefaultArtifactReducer +import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.EventMetadata +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.events.types.StageId +import com.correx.core.inference.InferenceRepository +import com.correx.core.inference.InferenceState +import com.correx.core.journal.DecisionJournalProjector +import com.correx.core.journal.DefaultDecisionJournalReducer +import com.correx.core.journal.DefaultDecisionJournalRepository +import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer +import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator +import com.correx.core.kernel.orchestration.OrchestrationProjector +import com.correx.core.kernel.orchestration.OrchestrationRepository +import com.correx.core.kernel.orchestration.OrchestratorEngines +import com.correx.core.kernel.orchestration.OrchestratorRepositories +import com.correx.core.kernel.retry.DefaultRetryCoordinator +import com.correx.core.risk.DefaultRiskAssessor +import com.correx.core.sessions.DefaultSessionRepository +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.sessions.projections.replay.EventReplayer +import com.correx.core.utils.TypeId +import com.correx.core.validation.pipeline.ValidationPipeline +import com.correx.infrastructure.persistence.InMemoryEventStore +import com.correx.infrastructure.persistence.artifact.LiveArtifactRepository +import com.correx.testing.fixtures.InferenceFixtures +import com.correx.testing.fixtures.context.ContextFixtures +import com.correx.testing.fixtures.transitions.TransitionFixtures +import com.correx.testing.kernel.MockSessionEventReplayer +import kotlinx.coroutines.runBlocking +import kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import java.util.UUID + +class RehydrateTest { + + private val sessionId: SessionId = TypeId("session-rehydrate") + private val artifactId: ArtifactId = TypeId("artifact-001") + private val artifactContent = """{"status":"success"}""" + + private val eventStore = InMemoryEventStore() + + private val fakeArtifactStore: ArtifactStore = object : ArtifactStore { + override suspend fun put(bytes: ByteArray): ArtifactId = artifactId + override suspend fun get(id: ArtifactId): ByteArray? = + if (id == artifactId) artifactContent.toByteArray(Charsets.UTF_8) else null + override suspend fun flushBefore(commit: suspend () -> Unit) = commit() + } + + private val orchestrationRepository = OrchestrationRepository( + DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())), + ) + + private val inferenceRepository = InferenceRepository( + object : EventReplayer { + override fun rebuild(sessionId: SessionId) = InferenceState() + }, + ) + + private val approvalRepository = DefaultApprovalRepository( + DefaultEventReplayer(eventStore, ApprovalProjector(DefaultApprovalReducer())), + ) + + private val decisionJournalRepository = DefaultDecisionJournalRepository( + DefaultEventReplayer(eventStore, DecisionJournalProjector(DefaultDecisionJournalReducer())), + ) + + private val repositories = OrchestratorRepositories( + eventStore = eventStore, + inferenceRepository = inferenceRepository, + orchestrationRepository = orchestrationRepository, + sessionRepository = DefaultSessionRepository(MockSessionEventReplayer()), + artifactRepository = LiveArtifactRepository(eventStore, DefaultArtifactReducer()), + approvalRepository = approvalRepository, + ) + + private val engines = OrchestratorEngines( + transitionResolver = TransitionFixtures.simpleResolver(), + contextPackBuilder = ContextFixtures.simpleBuilder(), + inferenceRouter = InferenceFixtures.fixedRouter(), + validationPipeline = ValidationPipeline(validators = emptyList()), + approvalEngine = DefaultApprovalEngine(), + riskAssessor = DefaultRiskAssessor(), + ) + + private val orchestrator = DefaultSessionOrchestrator( + repositories = repositories, + engines = engines, + retryCoordinator = DefaultRetryCoordinator(eventStore), + artifactStore = fakeArtifactStore, + decisionJournalRepository = decisionJournalRepository, + ) + + private suspend fun appendValidatedEvent() { + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = ArtifactValidatedEvent( + artifactId = artifactId, + sessionId = sessionId, + stageId = StageId("stage-1"), + ), + ), + ) + } + + @Test + fun `rehydrate populates artifactContentCache from ArtifactValidatedEvent`(): Unit = runBlocking { + appendValidatedEvent() + + orchestrator.rehydrate(sessionId) + + val cached = orchestrator.validatedArtifactContent(sessionId, artifactId) + assertEquals(artifactContent, cached) + } + + @Test + fun `rehydrate is a no-op when there are no ArtifactValidatedEvents`(): Unit = runBlocking { + orchestrator.rehydrate(sessionId) + + val cached = orchestrator.validatedArtifactContent(sessionId, artifactId) + assertEquals(null, cached) + } +}