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 a88df406..cb2fa0c4 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,10 +16,15 @@ import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.SessionId import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.withContext +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.jsonObject import org.slf4j.LoggerFactory private val log = LoggerFactory.getLogger("StreamQueries") +private val lenientJson = Json { ignoreUnknownKeys = true; isLenient = true } + /** * Read-only / snapshot queries served over the global stream: artifact listings and config * get/update. Kept separate from [GlobalStreamHandler] (which owns the connection lifecycle and @@ -64,7 +69,13 @@ class StreamQueries(private val module: ServerModule) { log.warn("artifact content missing for referenced hash={} (evicted/compacted?)", hash.value) return null } - return bytes.toString(Charsets.UTF_8).takeIf { it.isNotBlank() } + val rawString = bytes.toString(Charsets.UTF_8).takeIf { it.isNotBlank() } ?: return null + return dereferenceEnvelope(lenientJson, rawString) { innerHash -> + runCatching { withContext(Dispatchers.IO) { module.artifactStore.get(innerHash) } } + .getOrNull() + ?.toString(Charsets.UTF_8) + ?.takeIf { it.isNotBlank() } + } ?: rawString } /** Current editable config as a snapshot frame. [error] / [restartRequired] are set by updates. */ @@ -89,6 +100,36 @@ class StreamQueries(private val module: ServerModule) { } companion object { + /** + * If [rawContent] is a `FileWrittenArtifact` envelope (JSON object with both "path" and + * "contentHash" string fields), resolves the real content via [lookup] on the inner hash. + * Returns null when [rawContent] is not an envelope or [lookup] returns null (caller falls + * back to [rawContent]). Non-throwing — JSON parse failure silently returns null. + */ + internal suspend fun dereferenceEnvelope( + json: Json, + rawContent: String, + lookup: suspend (ArtifactId) -> String?, + ): String? { + val obj = runCatching { + json.parseToJsonElement(rawContent).jsonObject + }.getOrNull() ?: return null + + val innerHash = (obj["contentHash"] as? JsonPrimitive)?.takeIf { it.isString }?.content + ?.takeIf { it.isNotBlank() } ?: return null + val path = (obj["path"] as? JsonPrimitive)?.takeIf { it.isString }?.content + if (path == null) return null + + val resolved = lookup(ArtifactId(innerHash)) + if (resolved == null) { + log.warn( + "FileWrittenArtifact envelope references missing inner hash={} (evicted?)", + innerHash, + ) + } + return resolved + } + /** * 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 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 index 2ad14d12..d35feaba 100644 --- 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 @@ -10,7 +10,9 @@ 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.coroutines.runBlocking import kotlinx.datetime.Clock +import kotlinx.serialization.json.Json import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Test import java.util.UUID @@ -55,4 +57,36 @@ class StreamQueriesPlanArtifactsTest { assertEquals("VALIDATED", plan.phase) assertEquals("analyst", plan.stageId) } + + @Test + fun `dereferenceEnvelope returns inner content when envelope is valid`(): Unit = runBlocking { + val innerHash = ArtifactId("abc123") + val envelope = """{"path":"/src/Foo.kt","contentHash":"abc123","size":42,"mode":"0644"}""" + val innerContent = "fun foo() = 42" + val result = StreamQueries.dereferenceEnvelope( + json = Json { ignoreUnknownKeys = true; isLenient = true }, + rawContent = envelope, + ) { id -> if (id == innerHash) innerContent else null } + assertEquals(innerContent, result) + } + + @Test + fun `dereferenceEnvelope returns null when inner hash is missing from CAS`(): Unit = runBlocking { + val envelope = """{"path":"/src/Foo.kt","contentHash":"missing","size":42,"mode":"0644"}""" + val result = StreamQueries.dereferenceEnvelope( + json = Json { ignoreUnknownKeys = true; isLenient = true }, + rawContent = envelope, + ) { null } + assertEquals(null, result) + } + + @Test + fun `dereferenceEnvelope returns null when path is a nested object`(): Unit = runBlocking { + val envelope = """{"path":{"nested":"object"},"contentHash":"abc123"}""" + val result = StreamQueries.dereferenceEnvelope( + json = Json { ignoreUnknownKeys = true; isLenient = true }, + rawContent = envelope, + ) { id -> if (id.value == "abc123") "content" else null } + assertEquals(null, result) + } }