fix(server,kernel,events): harden artifact read + restore workspace on reopen + cleanups
Follow-ups spotted while fixing the artifact-content bug: - #1 robustness: listArtifacts resolves content via runCatching, degrading to null instead of failing the whole listing on one bad CAS read (resolveContent helper) - #2 eviction visibility: WARN when a referenced artifact hash resolves to null (most likely evicted/compacted while still referenced by the event log) - #3 ordering trap: document on ArtifactCreated/ArtifactContentStored that 'Created' fires at validation, AFTER ContentStored at inference — the latent fold hazard - #4 reopen gap: SessionSnapshot now carries workspaceRoot (derived from SessionWorkspaceBound), so a reopened session restores its workspace label - #5 orchestrator headroom: isBackEdge externalised to a top-level function so DefaultSessionOrchestrator drops below the TooManyFunctions threshold (was at it) #6 (orphaned empty F-010-era artifacts) needs no code change — empty content is already normalised to null ('(no content stored)'); emission is prevented going forward by the shipped F-010/F-018/F-020 fixes.
This commit is contained in:
@@ -23,6 +23,7 @@ import com.correx.core.events.events.ArtifactCreatedEvent
|
|||||||
import com.correx.core.events.events.OrchestrationPausedEvent
|
import com.correx.core.events.events.OrchestrationPausedEvent
|
||||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||||
import com.correx.core.events.events.StageCompletedEvent
|
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.StageFailedEvent
|
||||||
import com.correx.core.events.events.StoredEvent
|
import com.correx.core.events.events.StoredEvent
|
||||||
import com.correx.core.events.events.ToolExecutionCompletedEvent
|
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).
|
// Restore the output panel on reopen (otherwise the snapshot carries no output).
|
||||||
val lastOutput = resolveLastOutput(events)
|
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
|
// Re-register pending approvals so the ApprovalCoordinator can route responses
|
||||||
// from clients that connected after the ApprovalRequestedEvent was emitted.
|
// from clients that connected after the ApprovalRequestedEvent was emitted.
|
||||||
@@ -163,6 +167,7 @@ class SessionEventBridge(
|
|||||||
recentEvents = recentEvents,
|
recentEvents = recentEvents,
|
||||||
lastOutput = lastOutput,
|
lastOutput = lastOutput,
|
||||||
lastResponse = lastOutput,
|
lastResponse = lastOutput,
|
||||||
|
workspaceRoot = workspaceRoot,
|
||||||
lastSequence = lastGlobal,
|
lastSequence = lastGlobal,
|
||||||
lastSessionSequence = lastSession,
|
lastSessionSequence = lastSession,
|
||||||
))
|
))
|
||||||
|
|||||||
@@ -123,6 +123,9 @@ sealed interface ServerMessage {
|
|||||||
// output panel instead of showing blank until new activity arrives.
|
// output panel instead of showing blank until new activity arrives.
|
||||||
val lastOutput: String = "",
|
val lastOutput: String = "",
|
||||||
val lastResponse: 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 lastSequence: Long,
|
||||||
val lastSessionSequence: Long,
|
val lastSessionSequence: Long,
|
||||||
override val sequence: Long? = null,
|
override val sequence: Long? = null,
|
||||||
|
|||||||
@@ -16,6 +16,9 @@ import com.correx.core.events.types.ArtifactId
|
|||||||
import com.correx.core.events.types.SessionId
|
import com.correx.core.events.types.SessionId
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
import kotlinx.coroutines.withContext
|
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
|
* 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 {
|
suspend fun listArtifacts(sessionId: SessionId): ServerMessage.ArtifactList {
|
||||||
val events = withContext(Dispatchers.IO) { module.eventStore.read(sessionId) }
|
val events = withContext(Dispatchers.IO) { module.eventStore.read(sessionId) }
|
||||||
val artifacts = planArtifacts(events).map { plan ->
|
val artifacts = planArtifacts(events).map { plan ->
|
||||||
val content = plan.contentHash?.let {
|
|
||||||
withContext(Dispatchers.IO) { module.artifactStore.get(it) }?.toString(Charsets.UTF_8)
|
|
||||||
}
|
|
||||||
ArtifactSummaryDto(
|
ArtifactSummaryDto(
|
||||||
artifactId = plan.artifactId,
|
artifactId = plan.artifactId,
|
||||||
stageId = plan.stageId,
|
stageId = plan.stageId,
|
||||||
schemaVersion = plan.schemaVersion,
|
schemaVersion = plan.schemaVersion,
|
||||||
phase = plan.phase,
|
phase = plan.phase,
|
||||||
content = content?.takeIf { it.isNotBlank() },
|
content = resolveContent(plan.contentHash),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
return ServerMessage.ArtifactList(sessionId = sessionId, artifacts = artifacts)
|
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. */
|
/** Current editable config as a snapshot frame. [error] / [restartRequired] are set by updates. */
|
||||||
fun configSnapshot(error: String?, restartRequired: List<String>): ServerMessage.ConfigSnapshot {
|
fun configSnapshot(error: String?, restartRequired: List<String>): ServerMessage.ConfigSnapshot {
|
||||||
val fields = module.configService.snapshot().map { (field, value) ->
|
val fields = module.configService.snapshot().map { (field, value) ->
|
||||||
|
|||||||
@@ -396,6 +396,7 @@ func (m *Model) onSnapshot(msg protocol.ServerMessage) {
|
|||||||
}
|
}
|
||||||
sess.LastOutput = msg.LastOutput
|
sess.LastOutput = msg.LastOutput
|
||||||
sess.LastResponse = msg.LastResponse
|
sess.LastResponse = msg.LastResponse
|
||||||
|
sess.WorkspaceRoot = msg.WorkspaceRoot
|
||||||
for _, e := range msg.RecentEvents {
|
for _, e := range msg.RecentEvents {
|
||||||
sess.Events = append(sess.Events, EventEntry{formatTime(e.Timestamp), e.Type, e.Detail})
|
sess.Events = append(sess.Events, EventEntry{formatTime(e.Timestamp), e.Type, e.Detail})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -22,6 +22,13 @@ data class ArtifactValidatedEvent(
|
|||||||
val stageId: StageId,
|
val stageId: StageId,
|
||||||
) : EventPayload
|
) : 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
|
@Serializable
|
||||||
@SerialName("ArtifactCreated")
|
@SerialName("ArtifactCreated")
|
||||||
data class ArtifactCreatedEvent(
|
data class ArtifactCreatedEvent(
|
||||||
@@ -31,6 +38,10 @@ data class ArtifactCreatedEvent(
|
|||||||
val schemaVersion: Int,
|
val schemaVersion: Int,
|
||||||
) : EventPayload
|
) : 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
|
@Serializable
|
||||||
@SerialName("ArtifactContentStored")
|
@SerialName("ArtifactContentStored")
|
||||||
data class ArtifactContentStoredEvent(
|
data class ArtifactContentStoredEvent(
|
||||||
|
|||||||
+7
-5
@@ -14,6 +14,7 @@ import com.correx.core.events.events.ArtifactContentStoredEvent
|
|||||||
import com.correx.core.events.events.ArtifactValidatedEvent
|
import com.correx.core.events.events.ArtifactValidatedEvent
|
||||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||||
import com.correx.core.events.events.RefinementIterationEvent
|
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.events.TransitionExecutedEvent
|
||||||
import com.correx.core.events.orchestration.OrchestrationState
|
import com.correx.core.events.orchestration.OrchestrationState
|
||||||
import com.correx.core.events.types.ApprovalDecisionId
|
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
|
// 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
|
// event, so the guard is replay-deterministic. Exceeding the cap escalates to a
|
||||||
// terminal failure instead of looping forever.
|
// 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 cycleKey = "${ctx.currentStageId.value}->${nextStageId.value}"
|
||||||
val maxIterations = ctx.graph.stages[nextStageId]?.maxRetries ?: DEFAULT_MAX_REFINEMENT
|
val maxIterations = ctx.graph.stages[nextStageId]?.maxRetries ?: DEFAULT_MAX_REFINEMENT
|
||||||
val iteration = (orchestrationRepository.getState(ctx.sessionId).refinementIterations[cycleKey] ?: 0) + 1
|
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(
|
private fun ExecutionContext.enrich() = EnrichedExecutionContext(
|
||||||
graph, sessionId, stageCount, currentStageId, config,
|
graph, sessionId, stageCount, currentStageId, config,
|
||||||
session = repositories.sessionRepository.getSession(sessionId),
|
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<StoredEvent>, target: StageId): Boolean =
|
||||||
|
events.any { (it.payload as? TransitionExecutedEvent)?.to == target }
|
||||||
|
|
||||||
private sealed class StepResult {
|
private sealed class StepResult {
|
||||||
data class Continue(val ctx: EnrichedExecutionContext) : StepResult()
|
data class Continue(val ctx: EnrichedExecutionContext) : StepResult()
|
||||||
data class Terminal(val result: WorkflowResult) : StepResult()
|
data class Terminal(val result: WorkflowResult) : StepResult()
|
||||||
|
|||||||
Reference in New Issue
Block a user