feat(research): dedicated SourceFetched + LowQualityExtraction events (D)

- promote research fetch quality + content_sha256 from ToolExecutionCompletedEvent
  metadata to first-class SourceFetchedEvent / LowQualityExtractionEvent (registered
  + serialization test); existing metadata write kept intact (additive)
- emitted from SandboxedToolExecutor on research-fetch tool completion, guarded on
  the url/content_sha256/quality markers; reuses the extractor's LOW_QUALITY marker

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/events/src/main/kotlin/com/correx/core/events/events/ResearchSourceEvents.kt

@@ -0,0 +1,41 @@
+package com.correx.core.events.events
+
+import com.correx.core.events.types.SessionId
+import com.correx.core.events.types.StageId
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+
+/**
+ * A research T2 fetch completed and produced content at [contentSha256] (research-workflow-spec §3/§5).
+ *
+ * Promotes the fetch-quality + content-hash that previously lived only inside
+ * [ToolExecutionCompletedEvent] metadata into a first-class event, so "what sources were fetched"
+ * is queryable and renderable directly from the journal. Additive: the metadata write is retained
+ * for existing consumers. [quality] mirrors the extractor's quality marker as recorded in metadata.
+ */
+@Serializable
+@SerialName("SourceFetched")
+data class SourceFetchedEvent(
+    val sessionId: SessionId,
+    val stageId: StageId,
+    val url: String,
+    val contentSha256: String,
+    val quality: String,
+    val byteCount: Int = 0,
+) : EventPayload
+
+/**
+ * An extraction was retained but flagged low-quality — below the quality bar (research-workflow-spec §5)
+ * — for operator visibility. Emitted alongside [SourceFetchedEvent] when the fetch cleared rejection
+ * but fell under the minimum-content threshold (JS-rendered SPA, paywall); the content is still kept,
+ * the operator is simply warned that synthesis may want to route around it.
+ */
+@Serializable
+@SerialName("LowQualityExtraction")
+data class LowQualityExtractionEvent(
+    val sessionId: SessionId,
+    val stageId: StageId,
+    val url: String,
+    val contentSha256: String,
+    val reason: String,
+) : EventPayload

core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt

@@ -31,6 +31,7 @@ import com.correx.core.events.events.IdeaDiscardedEvent
 import com.correx.core.events.events.JournalCompactedEvent
 import com.correx.core.events.events.FileWrittenEvent
 import com.correx.core.events.events.L3MemoryRetrievedEvent
+import com.correx.core.events.events.LowQualityExtractionEvent
 import com.correx.core.events.events.InferenceCompletedEvent
 import com.correx.core.events.events.InitialIntentEvent
 import com.correx.core.events.events.InferenceFailedEvent
@@ -46,6 +47,7 @@ import com.correx.core.events.events.RepoMapComputedEvent
 import com.correx.core.events.events.RetryAttemptedEvent
 import com.correx.core.events.events.WorkspaceStateObservedEvent
 import com.correx.core.events.events.RiskAssessedEvent
+import com.correx.core.events.events.SourceFetchedEvent
 import com.correx.core.events.events.StageCheckpointFailedEvent
 import com.correx.core.events.events.StageCheckpointPassedEvent
 import com.correx.core.events.events.StageCompletedEvent
@@ -137,6 +139,8 @@ val eventModule = SerializersModule {
         subclass(StageCheckpointPassedEvent::class)
         subclass(StageCheckpointFailedEvent::class)
         subclass(PossibleContradictionFlaggedEvent::class)
+        subclass(SourceFetchedEvent::class)
+        subclass(LowQualityExtractionEvent::class)
     }
 }
 

core/events/src/test/kotlin/com/correx/core/events/serialization/ResearchSourceEventSerializationTest.kt

@@ -0,0 +1,47 @@
+package com.correx.core.events.serialization
+
+import com.correx.core.events.events.EventPayload
+import com.correx.core.events.events.LowQualityExtractionEvent
+import com.correx.core.events.events.SourceFetchedEvent
+import com.correx.core.events.types.SessionId
+import com.correx.core.events.types.StageId
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class ResearchSourceEventSerializationTest {
+
+    private val sessionId = SessionId("s")
+    private val stageId = StageId("st")
+
+    @Test
+    fun `SourceFetchedEvent round-trips as polymorphic EventPayload`() {
+        val sample: EventPayload = SourceFetchedEvent(
+            sessionId = sessionId,
+            stageId = stageId,
+            url = "https://example.com/a",
+            contentSha256 = "abc123",
+            quality = "OK",
+            byteCount = 4096,
+        )
+        val encoded = eventJson.encodeToString(EventPayload.serializer(), sample)
+        assertTrue(encoded.contains("\"type\":\"SourceFetched\""), "SerialName must be present: $encoded")
+        val decoded = eventJson.decodeFromString(EventPayload.serializer(), encoded)
+        assertEquals(sample, decoded)
+    }
+
+    @Test
+    fun `LowQualityExtractionEvent round-trips as polymorphic EventPayload`() {
+        val sample: EventPayload = LowQualityExtractionEvent(
+            sessionId = sessionId,
+            stageId = stageId,
+            url = "https://example.com/spa",
+            contentSha256 = "def456",
+            reason = "extracted 12 chars, below minimum 200",
+        )
+        val encoded = eventJson.encodeToString(EventPayload.serializer(), sample)
+        assertTrue(encoded.contains("\"type\":\"LowQualityExtraction\""), "SerialName must be present: $encoded")
+        val decoded = eventJson.decodeFromString(EventPayload.serializer(), encoded)
+        assertEquals(sample, decoded)
+    }
+}

infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutor.kt

@@ -4,12 +4,15 @@ import com.correx.core.artifactstore.ArtifactStore
 import com.correx.core.events.EventDispatcher
 import com.correx.core.events.events.EventPayload
 import com.correx.core.events.events.FileWrittenEvent
+import com.correx.core.events.events.LowQualityExtractionEvent
+import com.correx.core.events.events.SourceFetchedEvent
 import com.correx.core.events.events.ToolExecutionCompletedEvent
 import com.correx.core.events.events.ToolExecutionFailedEvent
 import com.correx.core.events.events.ToolExecutionStartedEvent
 import com.correx.core.events.events.ToolReceipt
 import com.correx.core.events.events.ToolRequest
 import com.correx.core.events.types.SessionId
+import com.correx.core.events.types.StageId
 import com.correx.core.events.types.ToolInvocationId
 import com.correx.core.tools.contract.FileAffectingTool
 import com.correx.core.tools.contract.Tool
@@ -82,6 +85,7 @@ class SandboxedToolExecutor(
                 restoreOrClean(backupMap, success = true)
                 cleanWorkingDir(workingDir)
                 emitCompleted(sessionId, invocationId, toolName, result, tool, affectedPaths, durationMs(), diff)
+                emitResearchSourceEvents(sessionId, request.stageId, result)
                 emitFileMutations(sessionId, invocationId, affectedPaths, preImages)
                 result
             }
@@ -183,6 +187,37 @@ class SandboxedToolExecutor(
         )
     }
 
+    /**
+     * Promotes the research fetch's quality + content-hash (BACKLOG §D) from the generic
+     * [ToolExecutionCompletedEvent] metadata into first-class events, additively: the metadata write
+     * above is left intact for existing consumers. Only fires for results that carry the fetch
+     * markers ([url] + [content_sha256] + [quality]), so non-fetch tools are untouched. The
+     * low-quality threshold reuses the extractor's own marker — [ExtractionQuality.LOW_QUALITY], whose
+     * name is what [WebFetchTool] writes into `quality` — rather than inventing a new policy here.
+     */
+    private suspend fun emitResearchSourceEvents(
+        sessionId: SessionId,
+        stageId: StageId,
+        result: ToolResult.Success,
+    ) {
+        val md = result.metadata
+        val url = md["url"]
+        val contentSha256 = md["content_sha256"]
+        val quality = md["quality"]
+        if (url == null || contentSha256 == null || quality == null) return
+        val byteCount = md["fetched_bytes"]?.toIntOrNull() ?: 0
+
+        emit(sessionId, SourceFetchedEvent(sessionId, stageId, url, contentSha256, quality, byteCount))
+
+        // Source of truth for the bar is the extractor (ExtractionQuality.LOW_QUALITY); its enum name
+        // is what WebFetchTool records in `quality`. Keep the marker as a literal to avoid a core ->
+        // infrastructure dependency inversion.
+        if (quality == LOW_QUALITY_MARKER) {
+            val reason = "extraction quality below the minimum-content bar ($byteCount bytes fetched)"
+            emit(sessionId, LowQualityExtractionEvent(sessionId, stageId, url, contentSha256, reason))
+        }
+    }
+
     @Suppress("MaxLineLength")
     private fun computeDiff(affectedPaths: Set<Path>, backupMap: Map<Path, Path>): String? {
         val diffs = mutableListOf<String>()
@@ -266,4 +301,9 @@ class SandboxedToolExecutor(
             )
         }
     }
+
+    private companion object {
+        /** Mirrors `ExtractionQuality.LOW_QUALITY.name` — the marker WebFetchTool writes into `quality`. */
+        const val LOW_QUALITY_MARKER = "LOW_QUALITY"
+    }
 }

infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorResearchSourceTest.kt

@@ -0,0 +1,127 @@
+package com.correx.infrastructure.tools
+
+import com.correx.core.approvals.Tier
+import com.correx.core.events.EventDispatcher
+import com.correx.core.events.events.EventPayload
+import com.correx.core.events.events.LowQualityExtractionEvent
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.SourceFetchedEvent
+import com.correx.core.events.events.StoredEvent
+import com.correx.core.events.events.ToolRequest
+import com.correx.core.events.stores.EventStore
+import com.correx.core.events.types.SessionId
+import com.correx.core.events.types.StageId
+import com.correx.core.events.types.ToolInvocationId
+import com.correx.core.tools.contract.ParamRole
+import com.correx.core.tools.contract.Tool
+import com.correx.core.tools.contract.ToolCapability
+import com.correx.core.tools.contract.ToolExecutor
+import com.correx.core.tools.contract.ToolResult
+import com.correx.core.tools.contract.ValidationResult
+import com.correx.core.tools.registry.ToolRegistry
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.json.JsonObject
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import java.nio.file.Files
+
+class SandboxedToolExecutorResearchSourceTest {
+
+    private class CapturingEventStore : EventStore {
+        val payloads = mutableListOf<EventPayload>()
+        override suspend fun append(event: NewEvent): StoredEvent {
+            payloads += event.payload
+            val seq = payloads.size.toLong()
+            return StoredEvent(event.metadata, seq, seq, event.payload)
+        }
+        override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = events.map { append(it) }
+        override fun read(sessionId: SessionId): List<StoredEvent> = emptyList()
+        override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = emptyList()
+        override fun lastSequence(sessionId: SessionId): Long? = null
+        override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
+        override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
+        override suspend fun lastGlobalSequence(): Long = payloads.size.toLong()
+        override fun allEvents(): Sequence<StoredEvent> = emptySequence()
+        override fun allSessionIds(): Set<SessionId> = emptySet()
+    }
+
+    private class SingleToolRegistry(private val tool: Tool) : ToolRegistry {
+        override fun resolve(name: String): Tool? = if (name == tool.name) tool else null
+        override fun all(): List<Tool> = listOf(tool)
+    }
+
+    /** Stands in for a research fetch: returns the same metadata shape WebFetchTool emits. */
+    private class FakeFetchTool(private val metadata: Map<String, String>) : Tool, ToolExecutor {
+        override val name: String = "web_fetch"
+        override val description: String = "fake"
+        override val tier: Tier = Tier.T2
+        override val requiredCapabilities: Set<ToolCapability> = setOf(ToolCapability.NETWORK_ACCESS)
+        override val paramRoles: Map<String, ParamRole> = mapOf("url" to ParamRole.NETWORK_TARGET)
+        override val parametersSchema: JsonObject = JsonObject(emptyMap())
+        override fun validateRequest(request: ToolRequest): ValidationResult = ValidationResult.Valid
+        override suspend fun execute(request: ToolRequest): ToolResult =
+            ToolResult.Success(request.invocationId, output = "md", metadata = metadata)
+    }
+
+    private fun request() = ToolRequest(
+        ToolInvocationId("inv"), SessionId("s"), StageId("st"), "web_fetch",
+        mapOf("url" to "https://example.com/a"),
+    )
+
+    private fun runFetch(metadata: Map<String, String>): CapturingEventStore {
+        val tool = FakeFetchTool(metadata)
+        val events = CapturingEventStore()
+        val exec = SandboxedToolExecutor(
+            delegate = tool,
+            registry = SingleToolRegistry(tool),
+            eventDispatcher = EventDispatcher(events),
+            workDir = Files.createTempDirectory("sbx-research"),
+        )
+        runBlocking { exec.execute(request()) }
+        return events
+    }
+
+    private fun fetchMetadata(quality: String) = mapOf(
+        "url" to "https://example.com/a",
+        "content_type" to "text/html",
+        "content_sha256" to "deadbeef",
+        "fetched_bytes" to "4096",
+        "extractor_version" to "html-md-1",
+        "quality" to quality,
+    )
+
+    @Test
+    fun `ok fetch emits SourceFetchedEvent and no low-quality event`() {
+        val events = runFetch(fetchMetadata("OK"))
+
+        val fetched = events.payloads.filterIsInstance<SourceFetchedEvent>().single()
+        assertEquals("https://example.com/a", fetched.url)
+        assertEquals("deadbeef", fetched.contentSha256)
+        assertEquals("OK", fetched.quality)
+        assertEquals(4096, fetched.byteCount)
+        assertTrue(events.payloads.filterIsInstance<LowQualityExtractionEvent>().isEmpty())
+    }
+
+    @Test
+    fun `low-quality fetch also emits LowQualityExtractionEvent`() {
+        val events = runFetch(fetchMetadata("LOW_QUALITY"))
+
+        assertEquals(1, events.payloads.filterIsInstance<SourceFetchedEvent>().size)
+        val low = events.payloads.filterIsInstance<LowQualityExtractionEvent>().single()
+        assertEquals("https://example.com/a", low.url)
+        assertEquals("deadbeef", low.contentSha256)
+        assertTrue(low.reason.isNotBlank())
+    }
+
+    @Test
+    fun `non-fetch tool result emits no research source events`() {
+        // No url/content_sha256/quality markers → the promotion path must stay silent.
+        val events = runFetch(mapOf("some" to "metadata"))
+
+        assertTrue(events.payloads.filterIsInstance<SourceFetchedEvent>().isEmpty())
+        assertTrue(events.payloads.filterIsInstance<LowQualityExtractionEvent>().isEmpty())
+    }
+}