diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/ResearchSourceEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/ResearchSourceEvents.kt new file mode 100644 index 00000000..48bfd0cb --- /dev/null +++ b/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 diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt index ddf42d40..0c215e13 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt +++ b/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) } } diff --git a/core/events/src/test/kotlin/com/correx/core/events/serialization/ResearchSourceEventSerializationTest.kt b/core/events/src/test/kotlin/com/correx/core/events/serialization/ResearchSourceEventSerializationTest.kt new file mode 100644 index 00000000..54263f1a --- /dev/null +++ b/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) + } +} diff --git a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutor.kt b/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutor.kt index 33927af4..f4d8cfd8 100644 --- a/infrastructure/tools/src/main/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutor.kt +++ b/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, backupMap: Map): String? { val diffs = mutableListOf() @@ -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" + } } diff --git a/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorResearchSourceTest.kt b/infrastructure/tools/src/test/kotlin/com/correx/infrastructure/tools/SandboxedToolExecutorResearchSourceTest.kt new file mode 100644 index 00000000..3e1e7195 --- /dev/null +++ b/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() + 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): List = events.map { append(it) } + override fun read(sessionId: SessionId): List = emptyList() + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = emptyList() + override fun lastSequence(sessionId: SessionId): Long? = null + override fun subscribe(sessionId: SessionId): Flow = emptyFlow() + override fun subscribeAll(): Flow = emptyFlow() + override suspend fun lastGlobalSequence(): Long = payloads.size.toLong() + override fun allEvents(): Sequence = emptySequence() + override fun allSessionIds(): Set = 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 = listOf(tool) + } + + /** Stands in for a research fetch: returns the same metadata shape WebFetchTool emits. */ + private class FakeFetchTool(private val metadata: Map) : Tool, ToolExecutor { + override val name: String = "web_fetch" + override val description: String = "fake" + override val tier: Tier = Tier.T2 + override val requiredCapabilities: Set = setOf(ToolCapability.NETWORK_ACCESS) + override val paramRoles: Map = 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): 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().single() + assertEquals("https://example.com/a", fetched.url) + assertEquals("deadbeef", fetched.contentSha256) + assertEquals("OK", fetched.quality) + assertEquals(4096, fetched.byteCount) + assertTrue(events.payloads.filterIsInstance().isEmpty()) + } + + @Test + fun `low-quality fetch also emits LowQualityExtractionEvent`() { + val events = runFetch(fetchMetadata("LOW_QUALITY")) + + assertEquals(1, events.payloads.filterIsInstance().size) + val low = events.payloads.filterIsInstance().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().isEmpty()) + assertTrue(events.payloads.filterIsInstance().isEmpty()) + } +}