kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kernel,server): rehydrate() + POST /sessions/{id}/resume

Browse Source
This commit is contained in:
kami
2026-06-08 10:13:33 +04:00
parent 5ad945ccb4
commit b830528d31
5 changed files with 192 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt
+19
View File
@@ -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)
apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt
+16
View File
@@ -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"]
core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt
+16
View File
@@ -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,
core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt
+1 -1
View File
@@ -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(),
) {
testing/integration/src/test/kotlin/RehydrateTest.kt
+140
View File
@@ -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<InferenceState> {
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)
}
}