fix(server): correct pause-reason labels (S7) + bind workspace on REST launch (S8)
S7: DomainEventMapper mapped every non-APPROVAL_PENDING pause (CLARIFICATION_PENDING, ABANDONED_STALE) to USER_REQUESTED, so clients rendered the wrong pause state. Added the two PauseReason enum values, mapped the real reason string, and gave the Go TUI distinct labels. S8: REST POST /sessions launched a run without resolving/recording a workspace, so the session had no SessionWorkspaceBoundEvent (path containment, project profile, grants had nothing to anchor to). Extracted the WS path's resolve+emit into ServerModule.bindWorkspace and called it from both launchers.
This commit is contained in:
@@ -7,6 +7,9 @@ import com.correx.apps.server.narration.NarrationSubscriber
|
|||||||
import com.correx.apps.server.registry.ProviderRegistry
|
import com.correx.apps.server.registry.ProviderRegistry
|
||||||
import com.correx.apps.server.registry.WorkflowRegistry
|
import com.correx.apps.server.registry.WorkflowRegistry
|
||||||
import com.correx.apps.server.workspace.WorkspaceResolver
|
import com.correx.apps.server.workspace.WorkspaceResolver
|
||||||
|
import com.correx.apps.server.workspace.WorkspaceResolution
|
||||||
|
import com.correx.core.events.events.SessionWorkspaceBoundEvent
|
||||||
|
import com.correx.core.kernel.orchestration.WorkspaceContext
|
||||||
import com.correx.core.approvals.ApprovalProjector
|
import com.correx.core.approvals.ApprovalProjector
|
||||||
import com.correx.core.approvals.DefaultApprovalReducer
|
import com.correx.core.approvals.DefaultApprovalReducer
|
||||||
import com.correx.core.approvals.DefaultApprovalRepository
|
import com.correx.core.approvals.DefaultApprovalRepository
|
||||||
@@ -296,6 +299,48 @@ class ServerModule(
|
|||||||
* run job is invisible to the guard and a second `resume()` is spuriously launched,
|
* run job is invisible to the guard and a second `resume()` is spuriously launched,
|
||||||
* duplicating inference calls (the root cause of the healthcheck script corruption).
|
* duplicating inference calls (the root cause of the healthcheck script corruption).
|
||||||
*/
|
*/
|
||||||
|
/**
|
||||||
|
* Resolves [workingDir] into a trusted workspace and records the decision as a
|
||||||
|
* [SessionWorkspaceBoundEvent] (invariant #9), returning the bound context or null when no
|
||||||
|
* resolver is configured. Shared by the WS StartSession path and the REST POST /sessions
|
||||||
|
* launcher so both anchor a session's workspace identically — the REST route previously skipped
|
||||||
|
* this entirely, leaving sessions with no bound workspace.
|
||||||
|
*/
|
||||||
|
suspend fun bindWorkspace(sessionId: SessionId, workingDir: String?): WorkspaceContext? {
|
||||||
|
val resolver = workspaceResolver ?: return null
|
||||||
|
val workspace = when (val resolution = withContext(Dispatchers.IO) { resolver.resolve(workingDir) }) {
|
||||||
|
is WorkspaceResolution.Bound -> {
|
||||||
|
log.info("workspace bound: session={} root={}", sessionId.value, resolution.workspace.workspaceRoot)
|
||||||
|
resolution.workspace
|
||||||
|
}
|
||||||
|
is WorkspaceResolution.Rejected -> {
|
||||||
|
log.warn(
|
||||||
|
"workspace rejected: session={} reason={} fallback={}",
|
||||||
|
sessionId.value, resolution.reason, resolution.fallback.workspaceRoot,
|
||||||
|
)
|
||||||
|
resolution.fallback
|
||||||
|
}
|
||||||
|
}
|
||||||
|
eventStore.append(
|
||||||
|
NewEvent(
|
||||||
|
metadata = EventMetadata(
|
||||||
|
eventId = EventId(java.util.UUID.randomUUID().toString()),
|
||||||
|
sessionId = sessionId,
|
||||||
|
timestamp = Clock.System.now(),
|
||||||
|
schemaVersion = 1,
|
||||||
|
causationId = null,
|
||||||
|
correlationId = null,
|
||||||
|
),
|
||||||
|
payload = SessionWorkspaceBoundEvent(
|
||||||
|
sessionId = sessionId,
|
||||||
|
workspaceRoot = workspace.workspaceRoot.toString(),
|
||||||
|
allowedPaths = workspace.allowedPaths.map { it.toString() },
|
||||||
|
),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
return workspace
|
||||||
|
}
|
||||||
|
|
||||||
fun launchSessionRun(
|
fun launchSessionRun(
|
||||||
sessionId: SessionId,
|
sessionId: SessionId,
|
||||||
graph: com.correx.core.transitions.graph.WorkflowGraph,
|
graph: com.correx.core.transitions.graph.WorkflowGraph,
|
||||||
|
|||||||
@@ -349,7 +349,12 @@ private fun mapOrchestrationPaused(
|
|||||||
seq: Long,
|
seq: Long,
|
||||||
sessionSequence: Long,
|
sessionSequence: Long,
|
||||||
): ServerMessage {
|
): ServerMessage {
|
||||||
val reason = if (p.reason == "APPROVAL_PENDING") PauseReason.APPROVAL_PENDING else PauseReason.USER_REQUESTED
|
val reason = when (p.reason) {
|
||||||
|
"APPROVAL_PENDING" -> PauseReason.APPROVAL_PENDING
|
||||||
|
"CLARIFICATION_PENDING" -> PauseReason.CLARIFICATION_PENDING
|
||||||
|
"ABANDONED_STALE" -> PauseReason.ABANDONED_STALE
|
||||||
|
else -> PauseReason.USER_REQUESTED
|
||||||
|
}
|
||||||
return ServerMessage.SessionPaused(
|
return ServerMessage.SessionPaused(
|
||||||
sessionId = p.sessionId,
|
sessionId = p.sessionId,
|
||||||
reason = reason,
|
reason = reason,
|
||||||
|
|||||||
@@ -61,6 +61,8 @@ data class ApprovalDto(
|
|||||||
@Serializable
|
@Serializable
|
||||||
enum class PauseReason {
|
enum class PauseReason {
|
||||||
APPROVAL_PENDING,
|
APPROVAL_PENDING,
|
||||||
|
CLARIFICATION_PENDING,
|
||||||
|
ABANDONED_STALE,
|
||||||
USER_REQUESTED,
|
USER_REQUESTED,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -121,10 +121,14 @@ private fun Route.startSessionRoute(module: ServerModule) {
|
|||||||
val graph = module.workflowRegistry.find(body.workflowId)
|
val graph = module.workflowRegistry.find(body.workflowId)
|
||||||
?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: ${body.workflowId}")
|
?: return@post call.respond(HttpStatusCode.BadRequest, "Unknown workflowId: ${body.workflowId}")
|
||||||
val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
|
val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
|
||||||
|
// Anchor the session's workspace (invariant #9) so path containment, project profile and
|
||||||
|
// grants resolve against a bound root. REST carries no cwd, so the resolver's default/
|
||||||
|
// fallback root is bound — parity with the WS StartSession path.
|
||||||
|
val workspace = module.bindWorkspace(sessionId, null)
|
||||||
body.intent?.takeIf { it.isNotBlank() }?.let { intent ->
|
body.intent?.takeIf { it.isNotBlank() }?.let { intent ->
|
||||||
EventDispatcher(module.eventStore).emit(InitialIntentEvent(sessionId, intent), sessionId)
|
EventDispatcher(module.eventStore).emit(InitialIntentEvent(sessionId, intent), sessionId)
|
||||||
}
|
}
|
||||||
module.launchSessionRun(sessionId, graph)
|
module.launchSessionRun(sessionId, graph, module.orchestrationConfig().copy(workspace = workspace))
|
||||||
call.respond(HttpStatusCode.Accepted, StartSessionResponse(sessionId.value))
|
call.respond(HttpStatusCode.Accepted, StartSessionResponse(sessionId.value))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -12,7 +12,6 @@ import com.correx.apps.server.protocol.ServerMessage
|
|||||||
import com.correx.apps.server.protocol.WorkflowDto
|
import com.correx.apps.server.protocol.WorkflowDto
|
||||||
import com.correx.apps.server.protocol.StageToolDecl
|
import com.correx.apps.server.protocol.StageToolDecl
|
||||||
import com.correx.apps.server.protocol.ToolDecl
|
import com.correx.apps.server.protocol.ToolDecl
|
||||||
import com.correx.apps.server.workspace.WorkspaceResolution
|
|
||||||
import com.correx.core.approvals.GRANT_LEDGER_SESSION_ID
|
import com.correx.core.approvals.GRANT_LEDGER_SESSION_ID
|
||||||
import com.correx.core.approvals.GrantScope
|
import com.correx.core.approvals.GrantScope
|
||||||
import com.correx.core.events.events.ApprovalGrantCreatedEvent
|
import com.correx.core.events.events.ApprovalGrantCreatedEvent
|
||||||
@@ -716,47 +715,10 @@ class GlobalStreamHandler(private val module: ServerModule) {
|
|||||||
val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
|
val sessionId: SessionId = TypeId(UUID.randomUUID().toString())
|
||||||
log.info("starting session={} workflow={}", sessionId.value, msg.workflowId)
|
log.info("starting session={} workflow={}", sessionId.value, msg.workflowId)
|
||||||
|
|
||||||
// Resolve workspace from the Hello-frame working directory (invariant #9: record the
|
// Resolve + record the workspace from the Hello-frame working directory (invariant #9:
|
||||||
// resolver's decision at handshake time as an event; replay reads the recorded fact).
|
// record the resolver's decision as an event; replay reads the recorded fact). Shared with
|
||||||
val resolvedWorkspace: WorkspaceContext? = module.workspaceResolver?.let { resolver ->
|
// the REST launcher via module.bindWorkspace.
|
||||||
val resolution = withContext(Dispatchers.IO) { resolver.resolve(workingDir) }
|
val resolvedWorkspace: WorkspaceContext? = module.bindWorkspace(sessionId, workingDir)
|
||||||
when (resolution) {
|
|
||||||
is WorkspaceResolution.Bound -> {
|
|
||||||
log.info("workspace bound: root={}", resolution.workspace.workspaceRoot)
|
|
||||||
resolution.workspace
|
|
||||||
}
|
|
||||||
is WorkspaceResolution.Rejected -> {
|
|
||||||
// Deliberate: a rejected client path still binds the resolver's fallback
|
|
||||||
// workspace. The fallback root is recorded in SessionWorkspaceBoundEvent so
|
|
||||||
// replay reproduces the same binding; the rejection reason is audit-logged
|
|
||||||
// here and does NOT become a field on the event (YAGNI).
|
|
||||||
log.warn(
|
|
||||||
"workspace rejected: session={} reason={} fallback={}",
|
|
||||||
sessionId.value, resolution.reason, resolution.fallback.workspaceRoot,
|
|
||||||
)
|
|
||||||
resolution.fallback
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Emit SessionWorkspaceBoundEvent when a workspace was resolved (invariant #9).
|
|
||||||
resolvedWorkspace?.let { ws ->
|
|
||||||
module.eventStore.append(NewEvent(
|
|
||||||
metadata = EventMetadata(
|
|
||||||
eventId = EventId(UUID.randomUUID().toString()),
|
|
||||||
sessionId = sessionId,
|
|
||||||
timestamp = Clock.System.now(),
|
|
||||||
schemaVersion = 1,
|
|
||||||
causationId = null,
|
|
||||||
correlationId = null,
|
|
||||||
),
|
|
||||||
payload = SessionWorkspaceBoundEvent(
|
|
||||||
sessionId = sessionId,
|
|
||||||
workspaceRoot = ws.workspaceRoot.toString(),
|
|
||||||
allowedPaths = ws.allowedPaths.map { it.toString() },
|
|
||||||
),
|
|
||||||
))
|
|
||||||
}
|
|
||||||
|
|
||||||
// Send StageToolManifest FIRST. If the WS is already closed, this throws and the
|
// Send StageToolManifest FIRST. If the WS is already closed, this throws and the
|
||||||
// orchestrator never launches — no silent data loss. The exception propagates to the
|
// orchestrator never launches — no silent data loss. The exception propagates to the
|
||||||
|
|||||||
@@ -120,8 +120,13 @@ func (m *Model) applyServer(msg protocol.ServerMessage) {
|
|||||||
}
|
}
|
||||||
case protocol.TypeSessionPaused:
|
case protocol.TypeSessionPaused:
|
||||||
label := "PAUSED"
|
label := "PAUSED"
|
||||||
if msg.Reason == "APPROVAL_PENDING" {
|
switch msg.Reason {
|
||||||
|
case "APPROVAL_PENDING":
|
||||||
label = "PAUSED awaiting approval"
|
label = "PAUSED awaiting approval"
|
||||||
|
case "CLARIFICATION_PENDING":
|
||||||
|
label = "PAUSED awaiting answer"
|
||||||
|
case "ABANDONED_STALE":
|
||||||
|
label = "PAUSED (stale)"
|
||||||
}
|
}
|
||||||
m.touch(msg.SessionID, label)
|
m.touch(msg.SessionID, label)
|
||||||
if s := m.session(msg.SessionID); s != nil {
|
if s := m.session(msg.SessionID); s != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user