diff --git a/apps/worker/src/main/java/com/orgmemory/worker/ingestion/DocumentProcessingEngine.java b/apps/worker/src/main/java/com/orgmemory/worker/ingestion/DocumentProcessingEngine.java index f1589895c..d5bdb9b2e 100644 --- a/apps/worker/src/main/java/com/orgmemory/worker/ingestion/DocumentProcessingEngine.java +++ b/apps/worker/src/main/java/com/orgmemory/worker/ingestion/DocumentProcessingEngine.java @@ -25,6 +25,7 @@ import com.orgmemory.graphrag.processing.ResolvedDocumentProcessingProfile; import com.orgmemory.integrations.graphrag.springai.JtokkitTextTokenizer; import com.orgmemory.integrations.graphrag.springai.SpringAiTextEmbeddingPort; +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.Optional; @@ -87,7 +88,10 @@ ProcessedSourceDocument process( DocumentParseRequest request, EmbeddingModel embeddingModel) { ParserSpec parser = parsers.route(request.suffix(), properties.parserId()); + long parseStartedAt = System.nanoTime(); var parsed = parser.parser().parse(request); + Duration parseDuration = + Duration.ofNanos(Math.max(0, System.nanoTime() - parseStartedAt)); long minimumChunks = Math.ceilDiv( (long) tokenizer.count(parsed.document().content()), properties.chunkSize()); @@ -106,6 +110,7 @@ ProcessedSourceDocument process( ChunkerOptions options = options(requestedChunker); Map requestedOptions = resolvedOptions(options); List output; + long chunkStartedAt = System.nanoTime(); try { output = chunkers.execute( requestedChunker, @@ -124,6 +129,11 @@ ProcessedSourceDocument process( new ChunkingRequest(parsed.document(), tokenizer, Optional.empty()), options); } + // Measured before the limit check so a rejected document is still measured work; the + // fallback path above is inside the same window on purpose, because a semantic chunker + // failing over to the recursive one is time this document actually spent chunking. + Duration chunkDuration = + Duration.ofNanos(Math.max(0, System.nanoTime() - chunkStartedAt)); if (output.size() > properties.maximumChunks()) { throw new RejectedSourceException( "CHUNK_LIMIT_EXCEEDED", @@ -144,7 +154,12 @@ ProcessedSourceDocument process( : Optional.empty(), resolvedOptions, parsed.document().contentSha256()); - return new ProcessedSourceDocument(parsed, output, profile); + return new ProcessedSourceDocument( + parsed, + output, + profile, + parseDuration, + chunkDuration); } private ChunkerOptions options(String chunkerId) { diff --git a/apps/worker/src/main/java/com/orgmemory/worker/ingestion/ProcessedSourceDocument.java b/apps/worker/src/main/java/com/orgmemory/worker/ingestion/ProcessedSourceDocument.java index e617bba98..d7a0b0be8 100644 --- a/apps/worker/src/main/java/com/orgmemory/worker/ingestion/ProcessedSourceDocument.java +++ b/apps/worker/src/main/java/com/orgmemory/worker/ingestion/ProcessedSourceDocument.java @@ -3,13 +3,27 @@ import com.orgmemory.graphrag.chunking.ChunkedText; import com.orgmemory.graphrag.parsing.DocumentParseResult; import com.orgmemory.graphrag.processing.ResolvedDocumentProcessingProfile; +import java.time.Duration; import java.util.List; import java.util.Objects; +/** + * @param parseDuration how long the routed parser took + * @param chunkDuration how long chunking took, including a semantic chunker's failover to the + * recursive one + * + *

Both are carried out of the engine rather than timed by the caller, because the caller + * makes one {@code process} call and cannot see where inside it parsing ended. Reporting the + * pair as one duration would hide which half a slow document spent its time in, and the two + * have unrelated causes — a parser handed a large scanned file, against a chunker falling + * back to a different algorithm. + */ record ProcessedSourceDocument( DocumentParseResult parseResult, List chunks, - ResolvedDocumentProcessingProfile profile) { + ResolvedDocumentProcessingProfile profile, + Duration parseDuration, + Duration chunkDuration) { ProcessedSourceDocument { Objects.requireNonNull(parseResult, "parseResult"); @@ -20,5 +34,10 @@ record ProcessedSourceDocument( "Processed document requires at least one chunk"); } Objects.requireNonNull(profile, "profile"); + Objects.requireNonNull(parseDuration, "parseDuration"); + Objects.requireNonNull(chunkDuration, "chunkDuration"); + if (parseDuration.isNegative() || chunkDuration.isNegative()) { + throw new IllegalArgumentException("durations must not be negative"); + } } } diff --git a/apps/worker/src/main/java/com/orgmemory/worker/ingestion/SourceIngestionProcessor.java b/apps/worker/src/main/java/com/orgmemory/worker/ingestion/SourceIngestionProcessor.java index 574ce1f22..94afe9a82 100644 --- a/apps/worker/src/main/java/com/orgmemory/worker/ingestion/SourceIngestionProcessor.java +++ b/apps/worker/src/main/java/com/orgmemory/worker/ingestion/SourceIngestionProcessor.java @@ -26,6 +26,7 @@ import com.orgmemory.core.knowledge.storage.ObjectKey; import com.orgmemory.core.knowledge.storage.ObjectStoragePort; import com.orgmemory.core.permission.AccessGate; +import com.orgmemory.graphrag.observability.GraphRagEventSink; import com.orgmemory.graphrag.parsing.DocumentParseRequest; import com.orgmemory.graphrag.parsing.DocumentParseResult; import java.io.IOException; @@ -37,8 +38,10 @@ import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.time.Duration; +import java.time.Instant; import java.util.HexFormat; import java.util.List; +import java.util.Objects; import java.util.Optional; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -46,6 +49,7 @@ import org.springframework.ai.embedding.EmbeddingModel; import org.springframework.ai.embedding.TokenCountBatchingStrategy; import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @Component @@ -62,7 +66,9 @@ class SourceIngestionProcessor { private final AiRouteResolver aiRoutes; private final SourceProcessingProperties properties; private final DocumentProcessingEngine processingEngine; + private final GraphRagEventSink events; + @Autowired SourceIngestionProcessor( SourceIngestionCoordinator coordinator, KnowledgeIngestionService ingestion, @@ -72,7 +78,33 @@ class SourceIngestionProcessor { ObjectProvider embeddingModels, AiRouteResolver aiRoutes, SourceProcessingProperties properties, - DocumentProcessingEngine processingEngine) { + DocumentProcessingEngine processingEngine, + ObjectProvider eventSinks) { + this( + coordinator, + ingestion, + publications, + embeddingProfiles, + objects, + embeddingModels, + aiRoutes, + properties, + processingEngine, + GraphRagEventSink.failureTolerant( + GraphRagEventSink.composite(eventSinks.orderedStream().toList()))); + } + + SourceIngestionProcessor( + SourceIngestionCoordinator coordinator, + KnowledgeIngestionService ingestion, + KnowledgeAssetPublicationService publications, + EmbeddingProfileRegistry embeddingProfiles, + ObjectStoragePort objects, + ObjectProvider embeddingModels, + AiRouteResolver aiRoutes, + SourceProcessingProperties properties, + DocumentProcessingEngine processingEngine, + GraphRagEventSink events) { this.coordinator = coordinator; this.ingestion = ingestion; this.publications = publications; @@ -82,6 +114,7 @@ class SourceIngestionProcessor { this.aiRoutes = aiRoutes; this.properties = properties; this.processingEngine = processingEngine; + this.events = Objects.requireNonNull(events, "events"); } void processNext() { @@ -89,6 +122,40 @@ void processNext() { .ifPresent(this::process); } + /** + * Reports the two ingestion stages that had no producer. + * + *

The revision status already moved through {@code PARSING} and {@code CHUNKING}, but a + * status says where a job is, not how long it stayed there — so a document that took four + * minutes to parse and one that took four seconds left the same trace. These are emitted + * from the same {@code jobId} the graph indexing stages use, so one upload reads as one + * operation across both processors. + */ + private void emitStage( + ClaimedSourceRevision claim, + GraphRagEventSink.Stage stage, + Duration duration, + int inputCount, + int outputCount) { + try { + events.emit(new GraphRagEventSink.GraphRagEvent( + claim.jobId(), + claim.organizationId(), + stage, + GraphRagEventSink.Outcome.SUCCEEDED, + duration, + inputCount, + outputCount, + null, + null, + null, + null, + Instant.now())); + } catch (RuntimeException ignoredTelemetryFailure) { + // Telemetry must never become an ingestion availability dependency. + } + } + private void process(ClaimedSourceRevision claim) { String failureStage = "VALIDATION"; Path temporaryFile = null; @@ -125,6 +192,18 @@ private void process(ClaimedSourceRevision claim) { Optional.empty()), embeddingModel); DocumentParseResult parsed = processed.parseResult(); + emitStage( + claim, + GraphRagEventSink.Stage.PARSE, + processed.parseDuration(), + 1, + parsed.document().blocks().size()); + emitStage( + claim, + GraphRagEventSink.Stage.CHUNK, + processed.chunkDuration(), + parsed.document().blocks().size(), + processed.chunks().size()); RawSourceRef raw = registerRawSource(claim, parsed); NormalizedRecordRef normalized = ingestion.normalize(new NormalizeRawSourceCommand( claim.organizationId(), diff --git a/apps/worker/src/test/java/com/orgmemory/worker/ingestion/DocumentProcessingEngineTests.java b/apps/worker/src/test/java/com/orgmemory/worker/ingestion/DocumentProcessingEngineTests.java index 83e1c642d..b9323102e 100644 --- a/apps/worker/src/test/java/com/orgmemory/worker/ingestion/DocumentProcessingEngineTests.java +++ b/apps/worker/src/test/java/com/orgmemory/worker/ingestion/DocumentProcessingEngineTests.java @@ -1,7 +1,9 @@ package com.orgmemory.worker.ingestion; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import com.orgmemory.graphrag.parsing.DocumentParseRequest; import java.nio.charset.StandardCharsets; @@ -48,6 +50,62 @@ void rejectsDefiniteChunkOverflowBeforeCallingSemanticEmbedding() { assertEquals("CHUNK_LIMIT_EXCEEDED", failure.code()); } + /** + * The caller makes one {@code process} call and cannot see where parsing ended, so the two + * measurements have to leave the engine separately or the pair collapses into one number + * that hides which half a slow document spent its time in. + * + *

This asserts the counts the two stage events report, not the accuracy of the timing. + * A timing assertion was tried and removed: on a fixture this small, starting the chunk + * clock before parsing still passed every bound the test could state, so it proved nothing + * it claimed to. The stage events themselves are proven end to end in + * {@code SourceIngestionPipelineIntegrationTests}. + */ + @Test + void carriesTheParseAndChunkMeasurementsOutSeparately() { + var engine = new DocumentProcessingEngine( + properties("passthrough", "fixed-token"), + new SpringAiDocumentParser()); + + ProcessedSourceDocument processed = engine.process( + new DocumentParseRequest( + "notes.txt", + "text/plain", + "one two three four five six seven eight" + .getBytes(StandardCharsets.UTF_8), + Optional.empty()), + new FailingEmbeddingModel()); + + assertFalse( + processed.parseResult().document().blocks().isEmpty(), + "PARSE reports blocks produced, so an empty list would make it report zero work"); + assertFalse( + processed.chunks().isEmpty(), + "CHUNK reports chunks produced"); + } + + private static SourceProcessingProperties properties( + String parserId, + String chunkerId) { + return new SourceProcessingProperties( + false, + Duration.ofSeconds(1), + "test-worker", + Duration.ofMinutes(1), + "test-pipeline", + parserId, + chunkerId, + "o200k_base", + "normalizer", + "fixture", + "fixture-model", + 64, + 8, + 0, + 64, + 1); + } + private static final class FailingEmbeddingModel implements EmbeddingModel { @Override diff --git a/apps/worker/src/test/java/com/orgmemory/worker/ingestion/SourceIngestionPipelineIntegrationTests.java b/apps/worker/src/test/java/com/orgmemory/worker/ingestion/SourceIngestionPipelineIntegrationTests.java index e531af94a..743f8587c 100644 --- a/apps/worker/src/test/java/com/orgmemory/worker/ingestion/SourceIngestionPipelineIntegrationTests.java +++ b/apps/worker/src/test/java/com/orgmemory/worker/ingestion/SourceIngestionPipelineIntegrationTests.java @@ -38,6 +38,7 @@ import com.orgmemory.core.knowledge.storage.StoredObject; import com.orgmemory.core.organization.CurrentActor; import com.orgmemory.core.permission.KnowledgeClassification; +import com.orgmemory.graphrag.observability.GraphRagEventSink; import java.io.ByteArrayInputStream; import java.nio.charset.StandardCharsets; import java.security.MessageDigest; @@ -129,6 +130,9 @@ class SourceIngestionPipelineIntegrationTests { @Autowired SourceIngestionProcessor processor; + @Autowired + RecordingGraphRagEventSink graphRagEvents; + @Autowired JdbcTemplate jdbc; @@ -414,6 +418,24 @@ INSERT INTO source_acl_snapshot_seals ( 1, staleAcl.evidence().size(), "The latest complete, sealed ACL generation remains authoritative when sync health is stale"); + + // Parsing and chunking had no producer at all, so an upload that spent minutes in either + // was indistinguishable from one that spent milliseconds — the revision status says where + // a job is, never how long it stayed there. + var parse = graphRagEvents.require(GraphRagEventSink.Stage.PARSE); + var chunk = graphRagEvents.require(GraphRagEventSink.Stage.CHUNK); + assertEquals(1, parse.inputCount(), "one source document entered parsing"); + assertTrue(parse.outputCount() > 0, "parsing reports the blocks it produced"); + assertEquals( + parse.outputCount(), + chunk.inputCount(), + "chunking consumes exactly the blocks parsing produced"); + assertTrue(chunk.outputCount() > 0, "chunking reports the chunks it produced"); + assertEquals( + parse.operationId(), + chunk.operationId(), + "both stages belong to the one job, so an upload reads as one operation"); + assertEquals(ORGANIZATION_ID, parse.organizationId()); } @Test @@ -559,6 +581,28 @@ private static float[] embedding() { return values; } + static final class RecordingGraphRagEventSink implements GraphRagEventSink { + + private final List received = + java.util.Collections.synchronizedList(new java.util.ArrayList<>()); + + @Override + public void emit(GraphRagEvent event) { + received.add(event); + } + + GraphRagEvent require(Stage stage) { + synchronized (received) { + return received.stream() + .filter(event -> event.stage() == stage) + .findFirst() + .orElseThrow(() -> new AssertionError( + "no " + stage + " event was emitted; saw " + + received.stream().map(GraphRagEvent::stage).toList())); + } + } + } + @TestConfiguration(proxyBeanMethods = false) @ComponentScan( basePackageClasses = SourceUploadService.class, @@ -568,6 +612,16 @@ private static float[] embedding() { pattern = "com\\.orgmemory\\.core\\.knowledge\\.Source(UploadService|UploadRegistrationService|QueryService)")) static class UploadTestConfiguration { + /** + * Contributed as an ordinary sink so the processor composes it exactly as it composes + * the real backends. A recorder injected any other way would prove the emit call and + * not the wiring, and the wiring is what was missing. + */ + @Bean + RecordingGraphRagEventSink recordingGraphRagEventSink() { + return new RecordingGraphRagEventSink(); + } + @Bean @Primary AiRouteResolver testAiRouteResolver() { diff --git a/docs/increments/active/2026-07-29-observability-pipeline/plan.md b/docs/increments/active/2026-07-29-observability-pipeline/plan.md index e01a766c1..96b8584d2 100644 --- a/docs/increments/active/2026-07-29-observability-pipeline/plan.md +++ b/docs/increments/active/2026-07-29-observability-pipeline/plan.md @@ -120,8 +120,14 @@ Depends on the composite sink merged in PR #132. second round's model time alone. Nothing is emitted when the profile disables gleaning, because a zero would claim a round never configured to run. -- [ ] `PARSE` and `CHUNK` need a sink in the ingestion pipeline, which holds - none today. +- [x] `PARSE` and `CHUNK`. The ingestion pipeline held no sink at all, so an + upload that spent minutes parsing was indistinguishable from one that spent + milliseconds — the revision status moved through `PARSING` and `CHUNKING`, + but a status says where a job is, not how long it stayed there. Both are + emitted under the same `jobId` the graph indexing stages use, so one upload + is one operation across two processors. The engine measures each window and + carries both out, because its caller makes one `process` call and cannot + see where parsing ended. - [ ] Deletion and rebuild: missing from the enum entirely, and the runbook requires a drill for it. Decide separately. - [ ] Extraction cost. `ExtractionRoundMetrics` already carries diff --git a/docs/specs/domains/secure-graph-rag.md b/docs/specs/domains/secure-graph-rag.md index 963e43dad..59870fb8a 100644 --- a/docs/specs/domains/secure-graph-rag.md +++ b/docs/specs/domains/secure-graph-rag.md @@ -147,9 +147,17 @@ Reconciled: `2026-07-29-observability-pipeline (de29c9e)`. environment-variable mapping so telemetry cannot acquire an unchosen destination. Spans carry `service.version` and `deployment.environment`. The worker samples traces in full and the API at 0.1. -- `Stage` declares fourteen values; production emits eleven. `PARSE`, `CHUNK` and - `GENERATE` have no producer, so no parsing, chunking or answer-generation - latency is reported. Deletion and rebuild have no stage. +- `Stage` declares fourteen values; production emits thirteen. `GENERATE` has no + producer, so no answer-generation latency is reported. Deletion and rebuild + have no stage. +- `PARSE` and `CHUNK` are emitted by source ingestion under the same `jobId` the + graph indexing stages use, so one upload reads as one operation across both + processors. `PARSE` reports one source document in and the canonical blocks + produced; `CHUNK` reports those blocks in and the chunks produced. The engine + measures both windows itself and carries them out on + `ProcessedSourceDocument`, because its caller makes one call and cannot see + where parsing ended. A semantic chunker's failover to the recursive chunker is + inside the chunk window, being time the document really spent chunking. - `GLEAN` reports the second extraction round separately from the first. It is emitted once per indexing job when the profile enables gleaning, counting chunks eligible against chunks that completed a gleaning round, so a token diff --git a/docs/tests/domains/secure-graph-rag.md b/docs/tests/domains/secure-graph-rag.md index 2a81a1068..270088acd 100644 --- a/docs/tests/domains/secure-graph-rag.md +++ b/docs/tests/domains/secure-graph-rag.md @@ -67,6 +67,13 @@ Reconciled: `2026-07-29-observability-pipeline (de29c9e)`. selected and what the model saw, and that consolidating two copies of one grounding reports nothing dropped, so deduplication cannot be mistaken for truncation. +- The ingestion pipeline integration test proves `PARSE` and `CHUNK` reach a sink + contributed as an ordinary bean — so the composition is exercised rather than + the emit call alone — that chunking consumes exactly the blocks parsing + produced, and that both carry the one job identifier. The engine unit test + covers only the counts those events report: a timing assertion was tried and + removed because on a small fixture, starting the chunk clock before parsing + still satisfied every bound the test could state. - Indexing telemetry tests prove the job emits extract, glean, merge, embed and publish in order; that `GLEAN` counts eligible chunks against chunks that completed a round, so a token guard declining one is visible; that its duration