From 5909030fc1f5a69c9e6b355d7f4667791f89871d Mon Sep 17 00:00:00 2001 From: kl3inIT Date: Thu, 30 Jul 2026 11:34:51 +0700 Subject: [PATCH] feat(ingestion): report parsing and chunking as stages MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Source ingestion held no event sink at all. The revision status already moved through PARSING and CHUNKING, but a status says where a job is, not how long it stayed there — so an upload that spent four minutes parsing a scanned PDF and one that spent four milliseconds on a text file left exactly the same trace. Both stages are emitted under the same jobId the graph indexing stages use, so one upload reads as one operation across two processors rather than as two unrelated halves. The engine measures each window and carries both durations out on ProcessedSourceDocument. Its caller makes one process call and cannot see where parsing ended, and reporting the pair as a single duration would hide which half a slow document spent its time in — the two have unrelated causes. A semantic chunker failing over to the recursive one stays inside the chunk window, being time the document really did spend chunking. The sink is composed the same way the indexing processor composes it, through failureTolerant(composite(...)), so ingestion telemetry fails the same way and never becomes an availability dependency. A timing assertion was written for the engine and removed. Starting the chunk clock before parsing still passed every bound it could state on a fixture that small, so it proved nothing it claimed. The stages are proven end to end in the pipeline integration test instead, against a real database, through a sink contributed as an ordinary bean so the wiring is exercised and not just the emit call. Co-Authored-By: Claude Opus 5 (1M context) --- .../ingestion/DocumentProcessingEngine.java | 17 +++- .../ingestion/ProcessedSourceDocument.java | 21 ++++- .../ingestion/SourceIngestionProcessor.java | 81 ++++++++++++++++++- .../DocumentProcessingEngineTests.java | 58 +++++++++++++ ...urceIngestionPipelineIntegrationTests.java | 54 +++++++++++++ .../2026-07-29-observability-pipeline/plan.md | 10 ++- docs/specs/domains/secure-graph-rag.md | 14 +++- docs/tests/domains/secure-graph-rag.md | 7 ++ 8 files changed, 254 insertions(+), 8 deletions(-) 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