kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): artifact viewer dropped content hash due to event ordering

Browse Source 
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
This commit is contained in:
kami
2026-06-10 12:49:11 +04:00
parent 90ba7244d5
commit 5f7b6f1f18
2 changed files with 113 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt
+55 -27
View File
@@ -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<String, ArtifactAcc>()
// 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<String, ArtifactId>()
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<StoredEvent>): List<ArtifactPlan> {
val accs = LinkedHashMap<String, ArtifactAcc>()
val lastInferenceByStage = HashMap<String, ArtifactId>()
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
apps/server/src/test/kotlin/com/correx/apps/server/ws/StreamQueriesPlanArtifactsTest.kt
+58
View File
@@ -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)
}
}