From 5f7b6f1f18f5b46ef1458a1272af88022dc03d64 Mon Sep 17 00:00:00 2001 From: kami Date: Wed, 10 Jun 2026 12:49:11 +0400 Subject: [PATCH] fix(server): artifact viewer dropped content hash due to event ordering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of empty artifact content (even for role_pipeline sessions with intact CAS data): ArtifactContentStored is emitted at inference time, BEFORE ArtifactCreated (which fires at validation). listArtifacts only applied the content hash via accs[id]?.let{} — a no-op because the artifact wasn't in the map yet — so the hash was silently dropped and every artifact showed blank. Verified against the live store (session 4679ed1e): all 5 artifacts had hash=none before, now resolve to their validated content (analysis 867B, design 1074B, impl_plan 966B, patch 574B, review_report 369B). - ArtifactContentStored now seeds the accumulator (getOrPut), order-independent - fold extracted to pure StreamQueries.planArtifacts for unit testing - regression test: content hash captured when ContentStored precedes Created - inference-output fallback retained for genuinely content-less stages --- .../correx/apps/server/ws/StreamQueries.kt | 82 +++++++++++++------ .../ws/StreamQueriesPlanArtifactsTest.kt | 58 +++++++++++++ 2 files changed, 113 insertions(+), 27 deletions(-) create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/ws/StreamQueriesPlanArtifactsTest.kt 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 73d42e54..c20e9095 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 @@ -11,6 +11,7 @@ import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.ArtifactValidatingEvent import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.StoredEvent import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.SessionId import kotlinx.coroutines.Dispatchers @@ -31,35 +32,15 @@ class StreamQueries(private val module: ServerModule) { */ suspend fun listArtifacts(sessionId: SessionId): ServerMessage.ArtifactList { val events = withContext(Dispatchers.IO) { module.eventStore.read(sessionId) } - val accs = LinkedHashMap() - // Per-stage raw inference output (last response), used as a content fallback for artifacts - // that never recorded a slot->hash mapping (ArtifactContentStoredEvent) — e.g. older sessions - // or stages where the validated-artifact link was lost. Lets the operator still see what the - // stage actually produced instead of a blank pane. - val lastInferenceByStage = HashMap() - for (event in events) { - when (val p = event.payload) { - is ArtifactCreatedEvent -> accs.getOrPut(p.artifactId.value) { ArtifactAcc() }.apply { - stageId = p.stageId.value - schemaVersion = p.schemaVersion - } - is ArtifactValidatingEvent -> accs[p.artifactId.value]?.let { it.phase = "VALIDATING" } - is ArtifactValidatedEvent -> accs[p.artifactId.value]?.let { it.phase = "VALIDATED" } - is ArtifactContentStoredEvent -> accs[p.artifactId.value]?.let { it.contentHash = p.contentHash } - is InferenceCompletedEvent -> lastInferenceByStage[p.stageId.value] = p.responseArtifactId - else -> Unit - } - } - val artifacts = accs.map { (id, acc) -> - val hash = acc.contentHash ?: lastInferenceByStage[acc.stageId] - val content = hash?.let { + val artifacts = planArtifacts(events).map { plan -> + val content = plan.contentHash?.let { withContext(Dispatchers.IO) { module.artifactStore.get(it) }?.toString(Charsets.UTF_8) } ArtifactSummaryDto( - artifactId = id, - stageId = acc.stageId, - schemaVersion = acc.schemaVersion, - phase = acc.phase, + artifactId = plan.artifactId, + stageId = plan.stageId, + schemaVersion = plan.schemaVersion, + phase = plan.phase, content = content?.takeIf { it.isNotBlank() }, ) } @@ -87,7 +68,54 @@ class StreamQueries(private val module: ServerModule) { is ConfigUpdateResult.Error -> configSnapshot(error = result.message, restartRequired = emptyList()) } - /** Mutable accumulator for folding a session's artifact events into one summary per artifact. */ + companion object { + /** + * Folds a session's events into one plan per artifact (creation order preserved). Pure — no + * CAS read — so it is unit-testable. [ArtifactContentStoredEvent] is emitted at inference + * time, BEFORE the [ArtifactCreatedEvent] that fires at validation, so BOTH seed the + * accumulator (`getOrPut`); folding only on creation silently dropped the content hash and + * blanked every artifact. As a fallback, an artifact with no stored hash uses its stage's + * last inference output so the operator still sees what the stage produced. + */ + internal fun planArtifacts(events: List): List { + val accs = LinkedHashMap() + val lastInferenceByStage = HashMap() + for (event in events) { + when (val p = event.payload) { + is ArtifactCreatedEvent -> accs.getOrPut(p.artifactId.value) { ArtifactAcc() }.apply { + stageId = p.stageId.value + schemaVersion = p.schemaVersion + } + is ArtifactValidatingEvent -> accs[p.artifactId.value]?.let { it.phase = "VALIDATING" } + is ArtifactValidatedEvent -> accs[p.artifactId.value]?.let { it.phase = "VALIDATED" } + is ArtifactContentStoredEvent -> + accs.getOrPut(p.artifactId.value) { ArtifactAcc() }.contentHash = p.contentHash + is InferenceCompletedEvent -> lastInferenceByStage[p.stageId.value] = p.responseArtifactId + else -> Unit + } + } + return accs.map { (id, acc) -> + ArtifactPlan( + artifactId = id, + stageId = acc.stageId, + schemaVersion = acc.schemaVersion, + phase = acc.phase, + contentHash = acc.contentHash ?: lastInferenceByStage[acc.stageId], + ) + } + } + } + + /** A folded artifact: identity/metadata plus the CAS hash to resolve its content from (if any). */ + internal data class ArtifactPlan( + val artifactId: String, + val stageId: String, + val schemaVersion: Int, + val phase: String, + val contentHash: ArtifactId?, + ) + + /** Mutable accumulator for folding a session's artifact events into one plan per artifact. */ private class ArtifactAcc { var stageId: String = "" var schemaVersion: Int = 0 diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/ws/StreamQueriesPlanArtifactsTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/ws/StreamQueriesPlanArtifactsTest.kt new file mode 100644 index 00000000..2ad14d12 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/ws/StreamQueriesPlanArtifactsTest.kt @@ -0,0 +1,58 @@ +package com.correx.apps.server.ws + +import com.correx.core.events.events.ArtifactContentStoredEvent +import com.correx.core.events.events.ArtifactCreatedEvent +import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.StoredEvent +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 kotlinx.datetime.Clock +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test +import java.util.UUID + +class StreamQueriesPlanArtifactsTest { + + private val sessionId = SessionId("s1") + private var seq = 0L + + private fun event(payload: EventPayload): StoredEvent = StoredEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + sequence = seq++, + sessionSequence = seq, + payload = payload, + ) + + // Regression: ArtifactContentStored is emitted at inference time, BEFORE ArtifactCreated (which + // fires at validation). The fold must still capture the content hash — folding only on creation + // dropped it and blanked every artifact in the viewer. + @Test + fun `captures content hash even when ArtifactContentStored precedes ArtifactCreated`() { + val hash = ArtifactId("83ef7566f2d82b5e") + val events = listOf( + event(ArtifactContentStoredEvent(ArtifactId("analysis"), hash, sessionId, StageId("analyst"))), + event(ArtifactCreatedEvent(ArtifactId("analysis"), sessionId, StageId("analyst"), schemaVersion = 1)), + event(ArtifactValidatedEvent(ArtifactId("analysis"), sessionId, StageId("analyst"))), + ) + + val plans = StreamQueries.planArtifacts(events) + + assertEquals(1, plans.size) + val plan = plans.single() + assertEquals("analysis", plan.artifactId) + assertEquals(hash, plan.contentHash) + assertEquals("VALIDATED", plan.phase) + assertEquals("analyst", plan.stageId) + } +}