diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt index 9492e5b0..183b0fe9 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt @@ -23,6 +23,7 @@ import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.StageCompletedEvent +import com.correx.core.events.events.SessionWorkspaceBoundEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.ToolExecutionCompletedEvent @@ -141,6 +142,9 @@ class SessionEventBridge( // Restore the output panel on reopen (otherwise the snapshot carries no output). val lastOutput = resolveLastOutput(events) + val workspaceRoot = events.lastOrNull { it.payload is SessionWorkspaceBoundEvent } + ?.let { (it.payload as SessionWorkspaceBoundEvent).workspaceRoot } + .orEmpty() // Re-register pending approvals so the ApprovalCoordinator can route responses // from clients that connected after the ApprovalRequestedEvent was emitted. @@ -163,6 +167,7 @@ class SessionEventBridge( recentEvents = recentEvents, lastOutput = lastOutput, lastResponse = lastOutput, + workspaceRoot = workspaceRoot, lastSequence = lastGlobal, lastSessionSequence = lastSession, )) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index b8395047..fa968451 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -123,6 +123,9 @@ sealed interface ServerMessage { // output panel instead of showing blank until new activity arrives. val lastOutput: String = "", val lastResponse: String = "", + // The session's bound workspace cwd, so a reopened session restores its workspace label + // (otherwise blank until a new workspace_bound, which only fires on a fresh start). + val workspaceRoot: String = "", val lastSequence: Long, val lastSessionSequence: Long, override val sequence: Long? = null, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt index c20e9095..a88df406 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt @@ -16,6 +16,9 @@ import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.SessionId import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext +import org.slf4j.LoggerFactory + +private val log = LoggerFactory.getLogger("StreamQueries") /** * Read-only / snapshot queries served over the global stream: artifact listings and config @@ -33,20 +36,37 @@ class StreamQueries(private val module: ServerModule) { suspend fun listArtifacts(sessionId: SessionId): ServerMessage.ArtifactList { val events = withContext(Dispatchers.IO) { module.eventStore.read(sessionId) } val artifacts = planArtifacts(events).map { plan -> - val content = plan.contentHash?.let { - withContext(Dispatchers.IO) { module.artifactStore.get(it) }?.toString(Charsets.UTF_8) - } ArtifactSummaryDto( artifactId = plan.artifactId, stageId = plan.stageId, schemaVersion = plan.schemaVersion, phase = plan.phase, - content = content?.takeIf { it.isNotBlank() }, + content = resolveContent(plan.contentHash), ) } return ServerMessage.ArtifactList(sessionId = sessionId, artifacts = artifacts) } + /** + * Resolves an artifact's UTF-8 content from CAS, degrading to null instead of throwing. A + * referenced hash that resolves to null is logged WARN — most likely the bytes were evicted by + * the CAS [com.correx.infrastructure.artifactscas.evict.Evictor] while the event log still + * references them, which would otherwise blank a single artifact with no signal. + */ + private suspend fun resolveContent(contentHash: ArtifactId?): String? { + val hash = contentHash ?: return null + val bytes = runCatching { withContext(Dispatchers.IO) { module.artifactStore.get(hash) } } + .getOrElse { + log.warn("artifact content read failed for hash={}: {}", hash.value, it.message) + null + } + if (bytes == null) { + log.warn("artifact content missing for referenced hash={} (evicted/compacted?)", hash.value) + return null + } + return bytes.toString(Charsets.UTF_8).takeIf { it.isNotBlank() } + } + /** Current editable config as a snapshot frame. [error] / [restartRequired] are set by updates. */ fun configSnapshot(error: String?, restartRequired: List): ServerMessage.ConfigSnapshot { val fields = module.configService.snapshot().map { (field, value) -> diff --git a/apps/tui-go/internal/app/server.go b/apps/tui-go/internal/app/server.go index 30aa5c98..b732d2cd 100644 --- a/apps/tui-go/internal/app/server.go +++ b/apps/tui-go/internal/app/server.go @@ -396,6 +396,7 @@ func (m *Model) onSnapshot(msg protocol.ServerMessage) { } sess.LastOutput = msg.LastOutput sess.LastResponse = msg.LastResponse + sess.WorkspaceRoot = msg.WorkspaceRoot for _, e := range msg.RecentEvents { sess.Events = append(sess.Events, EventEntry{formatTime(e.Timestamp), e.Type, e.Detail}) } diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/ArtifactEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/ArtifactEvents.kt index 8c271307..4360e469 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/ArtifactEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/ArtifactEvents.kt @@ -22,6 +22,13 @@ data class ArtifactValidatedEvent( val stageId: StageId, ) : EventPayload +/** + * ⚠️ ORDERING TRAP: despite the name, this is emitted at the *validation* phase — AFTER + * [ArtifactContentStoredEvent] (which fires at inference time). Any fold that derives per-artifact + * state must NOT assume `ArtifactCreated` is the first event for an [artifactId]; seed accumulators + * from whichever artifact event arrives first (see StreamQueries.planArtifacts, which was bitten by + * exactly this — the content hash arrives one event earlier). + */ @Serializable @SerialName("ArtifactCreated") data class ArtifactCreatedEvent( @@ -31,6 +38,10 @@ data class ArtifactCreatedEvent( val schemaVersion: Int, ) : EventPayload +/** + * Records the slot→CAS-hash mapping for a produced artifact. Emitted at inference time, so it + * precedes [ArtifactCreatedEvent] for the same [artifactId] (see the ordering note there). + */ @Serializable @SerialName("ArtifactContentStored") data class ArtifactContentStoredEvent( 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 38ac0277..a463725f 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 @@ -14,6 +14,7 @@ import com.correx.core.events.events.ArtifactContentStoredEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RefinementIterationEvent +import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.types.ApprovalDecisionId @@ -274,7 +275,7 @@ class DefaultSessionOrchestrator( // run, e.g. reviewer→implementer) increments a per-cycle counter recorded as an // event, so the guard is replay-deterministic. Exceeding the cap escalates to a // terminal failure instead of looping forever. - if (isBackEdge(ctx, nextStageId)) { + if (isBackEdge(repositories.eventStore.read(ctx.sessionId), nextStageId)) { val cycleKey = "${ctx.currentStageId.value}->${nextStageId.value}" val maxIterations = ctx.graph.stages[nextStageId]?.maxRetries ?: DEFAULT_MAX_REFINEMENT val iteration = (orchestrationRepository.getState(ctx.sessionId).refinementIterations[cycleKey] ?: 0) + 1 @@ -391,10 +392,6 @@ class DefaultSessionOrchestrator( } } - private fun isBackEdge(ctx: EnrichedExecutionContext, target: StageId): Boolean = - repositories.eventStore.read(ctx.sessionId) - .any { (it.payload as? TransitionExecutedEvent)?.to == target } - private fun ExecutionContext.enrich() = EnrichedExecutionContext( graph, sessionId, stageCount, currentStageId, config, session = repositories.sessionRepository.getSession(sessionId), @@ -402,6 +399,11 @@ class DefaultSessionOrchestrator( ) } +// A back-edge re-enters a stage already transitioned into this run (e.g. reviewer→implementer). +// Top-level (not a member) to keep the orchestrator off the TooManyFunctions threshold. +private fun isBackEdge(events: List, target: StageId): Boolean = + events.any { (it.payload as? TransitionExecutedEvent)?.to == target } + private sealed class StepResult { data class Continue(val ctx: EnrichedExecutionContext) : StepResult() data class Terminal(val result: WorkflowResult) : StepResult()