fix(server,kernel): freestyle survives kill/restart at every phase (B4)
Three gaps found by live QA (kill between planning-complete and plan lock):
- launchSessionResumeWithRehydrate never handed off to FreestyleDriver, so a
resumed freestyle session stranded at planning-complete with no plan lock
and no phase 2; now locks+runs when planning completed and no
ExecutionPlanLockedEvent exists yet.
- resume route 400'd on phase-2 sessions (workflowId freestyle-<sid> is
compiled, never registered); now recompiles the locked plan after
rehydrating the artifact cache.
- orchestrator.resume threw when currentStageId was terminal ('done') or
unknown to the graph; now returns WorkflowResult.Completed.
This commit is contained in:
@@ -15,6 +15,7 @@ import com.correx.core.config.ProjectProfileLoader
|
|||||||
import com.correx.core.events.events.ApprovalRequestedEvent
|
import com.correx.core.events.events.ApprovalRequestedEvent
|
||||||
import com.correx.core.events.events.EventMetadata
|
import com.correx.core.events.events.EventMetadata
|
||||||
import com.correx.core.events.events.NewEvent
|
import com.correx.core.events.events.NewEvent
|
||||||
|
import com.correx.core.events.events.ExecutionPlanLockedEvent
|
||||||
import com.correx.core.events.events.OperatorProfileBoundEvent
|
import com.correx.core.events.events.OperatorProfileBoundEvent
|
||||||
import com.correx.core.events.events.ProjectProfileBoundEvent
|
import com.correx.core.events.events.ProjectProfileBoundEvent
|
||||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||||
@@ -22,6 +23,7 @@ import com.correx.core.events.events.WorkflowFailedEvent
|
|||||||
import com.correx.core.events.stores.EventStore
|
import com.correx.core.events.stores.EventStore
|
||||||
import com.correx.core.events.types.EventId
|
import com.correx.core.events.types.EventId
|
||||||
import com.correx.core.events.types.SessionId
|
import com.correx.core.events.types.SessionId
|
||||||
|
import com.correx.core.kernel.execution.WorkflowResult
|
||||||
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
|
import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
|
||||||
import com.correx.core.kernel.orchestration.OrchestrationConfig
|
import com.correx.core.kernel.orchestration.OrchestrationConfig
|
||||||
import com.correx.core.kernel.orchestration.OrchestrationRepository
|
import com.correx.core.kernel.orchestration.OrchestrationRepository
|
||||||
@@ -263,6 +265,18 @@ class ServerModule(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Resolves the workflow graph for resuming a freestyle phase-2 session. Phase-2 graphs
|
||||||
|
* (`freestyle-<sessionId>`) are compiled from the locked plan, never registered in the
|
||||||
|
* TOML registry, so the resume route cannot find them there. Rehydrates the artifact
|
||||||
|
* cache first because the plan content lives in it after a restart.
|
||||||
|
*/
|
||||||
|
suspend fun freestyleResumeGraph(sessionId: SessionId, workflowId: String): com.correx.core.transitions.graph.WorkflowGraph? {
|
||||||
|
if (!workflowId.startsWith("freestyle-")) return null
|
||||||
|
orchestrator.rehydrate(sessionId)
|
||||||
|
return freestyleDriver?.compiledGraph(sessionId)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Bind the per-repo project profile (curated .correx/project.toml) as an event so stages,
|
* Bind the per-repo project profile (curated .correx/project.toml) as an event so stages,
|
||||||
* router chat triage, and replay read the recorded snapshot, never the live file
|
* router chat triage, and replay read the recorded snapshot, never the live file
|
||||||
@@ -318,7 +332,18 @@ class ServerModule(
|
|||||||
moduleScope.launch {
|
moduleScope.launch {
|
||||||
runCatching {
|
runCatching {
|
||||||
orchestrator.rehydrate(sessionId)
|
orchestrator.rehydrate(sessionId)
|
||||||
orchestrator.resume(sessionId, graph, orchestrationConfig())
|
val result = orchestrator.resume(sessionId, graph, orchestrationConfig())
|
||||||
|
// Freestyle handoff parity with launchSessionRun: a session killed between
|
||||||
|
// planning completion and plan lock would otherwise strand forever — the
|
||||||
|
// resume completes the planning graph but nothing ever locks the plan.
|
||||||
|
val planAlreadyLocked = eventStore.read(sessionId)
|
||||||
|
.any { it.payload is ExecutionPlanLockedEvent }
|
||||||
|
if (graph.id == "freestyle_planning" &&
|
||||||
|
result is WorkflowResult.Completed &&
|
||||||
|
!planAlreadyLocked
|
||||||
|
) {
|
||||||
|
freestyleDriver?.lockAndRun(sessionId)
|
||||||
|
}
|
||||||
}.onFailure { e ->
|
}.onFailure { e ->
|
||||||
log.error("resumeSession (rehydrate): session={} failed", sessionId.value, e)
|
log.error("resumeSession (rehydrate): session={} failed", sessionId.value, e)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -71,6 +71,18 @@ class FreestyleDriver(
|
|||||||
runPhase2(sessionId, graph, config)
|
runPhase2(sessionId, graph, config)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Recompiles the stored execution plan for resume-after-restart. Requires the
|
||||||
|
* orchestrator's artifact cache to be rehydrated first. Null when the plan is
|
||||||
|
* absent or no longer compiles — callers fall back to a route-level error.
|
||||||
|
*/
|
||||||
|
fun compiledGraph(sessionId: SessionId): WorkflowGraph? {
|
||||||
|
val json = planContent(sessionId) ?: return null
|
||||||
|
return runCatching { compiler.compile(json, "freestyle-${sessionId.value}") }
|
||||||
|
.onFailure { log.error("freestyle: plan recompile for resume failed: {}", it.message) }
|
||||||
|
.getOrNull()
|
||||||
|
}
|
||||||
|
|
||||||
private suspend fun emitRejected(sessionId: SessionId, reason: String, source: String) {
|
private suspend fun emitRejected(sessionId: SessionId, reason: String, source: String) {
|
||||||
eventStore.append(
|
eventStore.append(
|
||||||
NewEvent(
|
NewEvent(
|
||||||
|
|||||||
@@ -129,6 +129,7 @@ private fun Route.resumeSessionRoute(module: ServerModule) {
|
|||||||
val workflowId = orchState.workflowId.takeIf { it.isNotBlank() }
|
val workflowId = orchState.workflowId.takeIf { it.isNotBlank() }
|
||||||
?: return@post call.respond(HttpStatusCode.BadRequest, "Session has no workflowId")
|
?: return@post call.respond(HttpStatusCode.BadRequest, "Session has no workflowId")
|
||||||
val graph = module.workflowRegistry.find(workflowId)
|
val graph = module.workflowRegistry.find(workflowId)
|
||||||
|
?: module.freestyleResumeGraph(TypeId(id), workflowId)
|
||||||
?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: $workflowId")
|
?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: $workflowId")
|
||||||
module.launchSessionResumeWithRehydrate(TypeId(id), graph)
|
module.launchSessionResumeWithRehydrate(TypeId(id), graph)
|
||||||
call.respond(HttpStatusCode.Accepted)
|
call.respond(HttpStatusCode.Accepted)
|
||||||
|
|||||||
+9
@@ -219,6 +219,15 @@ class DefaultSessionOrchestrator(
|
|||||||
): WorkflowResult {
|
): WorkflowResult {
|
||||||
val stageId = orchestrationRepository.getState(sessionId).currentStageId
|
val stageId = orchestrationRepository.getState(sessionId).currentStageId
|
||||||
?: return WorkflowResult.Failed(sessionId, "resume: no currentStageId", retryExhausted = false)
|
?: return WorkflowResult.Failed(sessionId, "resume: no currentStageId", retryExhausted = false)
|
||||||
|
if (!graph.stages.containsKey(stageId)) {
|
||||||
|
// Terminal pseudo-stage ("done") or a stage from another graph — nothing to
|
||||||
|
// re-enter. Treat as already completed instead of throwing inside enterStage.
|
||||||
|
log.info(
|
||||||
|
"[Orchestrator] resume: session={} stage={} not in graph '{}' — already terminal",
|
||||||
|
sessionId.value, stageId.value, graph.id,
|
||||||
|
)
|
||||||
|
return WorkflowResult.Completed(sessionId, stageId)
|
||||||
|
}
|
||||||
log.info("[Orchestrator] resuming session={} workflow={} stage={}", sessionId.value, graph.id, stageId.value)
|
log.info("[Orchestrator] resuming session={} workflow={} stage={}", sessionId.value, graph.id, stageId.value)
|
||||||
val base = ExecutionContext(graph, sessionId, 0, stageId, config, null, null)
|
val base = ExecutionContext(graph, sessionId, 0, stageId, config, null, null)
|
||||||
val enriched = base.enrich()
|
val enriched = base.enrich()
|
||||||
|
|||||||
Reference in New Issue
Block a user