diff --git a/apps/worker/src/main/java/com/orgmemory/worker/graph/GraphIndexingProcessor.java b/apps/worker/src/main/java/com/orgmemory/worker/graph/GraphIndexingProcessor.java index 1c89412f4..c41dfbd45 100644 --- a/apps/worker/src/main/java/com/orgmemory/worker/graph/GraphIndexingProcessor.java +++ b/apps/worker/src/main/java/com/orgmemory/worker/graph/GraphIndexingProcessor.java @@ -12,7 +12,9 @@ import com.orgmemory.graphrag.indexing.LightRagEmbeddingPayloads; import com.orgmemory.graphrag.model.ContributionEmbedding; import com.orgmemory.graphrag.model.EntityContribution; +import com.orgmemory.graphrag.model.ExtractionDiagnostics; import com.orgmemory.graphrag.model.ExtractionProfile; +import com.orgmemory.graphrag.model.ExtractionRoundMetrics; import com.orgmemory.graphrag.model.FloatVector; import com.orgmemory.graphrag.model.RelationContribution; import com.orgmemory.graphrag.observability.GraphRagEventSink; @@ -128,6 +130,7 @@ private void process(ClaimedGraphIndex claim) { claim.chunks().size(), () -> extractChunks(claim, extractionProfile, extractor), List::size); + emitGleaning(claim, extractionProfile, extracted); var contributions = observed( claim, GraphRagEventSink.Stage.MERGE, @@ -239,6 +242,65 @@ private T observed( } } + /** + * Reports the second extraction round separately from the first. + * + *

Gleaning is a second model call per chunk that exists to recover entities the first + * round missed, and it is the part of extraction a profile can turn off or a token guard can + * decline. Folded into {@code EXTRACT} it was indistinguishable from the round that always + * runs, so a deployment could not tell gleaning working from gleaning silently not + * happening — and the extractor was already recording everything needed to say which. + * + *

The duration is aggregate model time across chunks, not wall clock: chunks glean + * concurrently, so this is what gleaning cost rather than how long it took. It is nested + * inside the {@code EXTRACT} wall clock rather than sequential with it, so stage durations + * for one job must not be summed. + * + *

Nothing is emitted when the profile disables gleaning. A zero-valued series would claim + * a round that was never configured to run. + */ + private void emitGleaning( + ClaimedGraphIndex claim, + ExtractionProfile profile, + List extracted) { + if (profile.maxGleaningRounds() <= 0) { + return; + } + long elapsedNanos = 0; + int completed = 0; + for (ExtractedChunk chunk : extracted) { + ExtractionDiagnostics diagnostics = chunk.result().diagnostics(); + if (diagnostics.gleaningOutcome() + != ExtractionDiagnostics.GleaningOutcome.COMPLETED) { + continue; + } + completed++; + for (ExtractionRoundMetrics round : diagnostics.rounds()) { + if (round.round() > 0) { + elapsedNanos = Math.addExact( + elapsedNanos, round.elapsed().toNanos()); + } + } + } + try { + events.emit(new GraphRagEventSink.GraphRagEvent( + claim.jobId(), + claim.organizationId(), + GraphRagEventSink.Stage.GLEAN, + GraphRagEventSink.Outcome.SUCCEEDED, + Duration.ofNanos(elapsedNanos), + extracted.size(), + completed, + null, + null, + null, + null, + Instant.now())); + } catch (RuntimeException ignoredTelemetryFailure) { + // Telemetry must never become an indexing availability dependency. + } + } + private void emit( ClaimedGraphIndex claim, GraphRagEventSink.Stage stage, diff --git a/apps/worker/src/test/java/com/orgmemory/worker/graph/GraphIndexingProcessorTests.java b/apps/worker/src/test/java/com/orgmemory/worker/graph/GraphIndexingProcessorTests.java index ee27552c8..240dd82fc 100644 --- a/apps/worker/src/test/java/com/orgmemory/worker/graph/GraphIndexingProcessorTests.java +++ b/apps/worker/src/test/java/com/orgmemory/worker/graph/GraphIndexingProcessorTests.java @@ -7,6 +7,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; @@ -27,8 +28,10 @@ import com.orgmemory.graphrag.extraction.LightRagExtractionPrompt; import com.orgmemory.graphrag.model.ExtractedEntity; import com.orgmemory.graphrag.model.ExtractedRelation; -import com.orgmemory.graphrag.model.ExtractionResult; +import com.orgmemory.graphrag.model.ExtractionDiagnostics; import com.orgmemory.graphrag.model.ExtractionProfile; +import com.orgmemory.graphrag.model.ExtractionResult; +import com.orgmemory.graphrag.model.ExtractionRoundMetrics; import com.orgmemory.graphrag.model.FloatVector; import com.orgmemory.graphrag.model.RelationOrientation; import com.orgmemory.graphrag.observability.GraphRagEventSink; @@ -162,10 +165,11 @@ void publishesOneAtomicProjectionAndCompletesTheDurableJob() { verify(coordinator, never()).fail(any(), any(), any(), any()); ArgumentCaptor emitted = ArgumentCaptor.forClass(GraphRagEventSink.GraphRagEvent.class); - verify(events, times(4)).emit(emitted.capture()); + verify(events, times(5)).emit(emitted.capture()); assertEquals( List.of( GraphRagEventSink.Stage.EXTRACT, + GraphRagEventSink.Stage.GLEAN, GraphRagEventSink.Stage.MERGE, GraphRagEventSink.Stage.EMBED, GraphRagEventSink.Stage.PUBLISH), @@ -174,6 +178,78 @@ void publishesOneAtomicProjectionAndCompletesTheDurableJob() { .toList()); } + /** + * Gleaning is the extraction round a profile can disable and a token guard can decline, and + * folding it into {@code EXTRACT} made gleaning working indistinguishable from gleaning + * silently not happening. The extractor was already recording which. + */ + @Test + void reportsHowManyChunksCompletedGleaningAndWhatItCost() { + GraphRagEventSink events = mock(GraphRagEventSink.class); + ClaimedGraphIndex claim = claim(List.of( + chunk(CHUNK_ID, 0, "OrgMemory builds secure retrieval.", null), + chunk(SECOND_CHUNK_ID, 1, "OrgMemory also builds retrieval.", null))); + java.util.concurrent.atomic.AtomicInteger call = + new java.util.concurrent.atomic.AtomicInteger(); + + processorFor(claim, events, request -> extraction( + request, + // Only the first chunk gleans; the second is declined by the token guard, which + // is the case a folded-in stage could not distinguish from gleaning being off. + call.getAndIncrement() == 0 + ? new ExtractionDiagnostics( + List.of( + round(0, Duration.ofMillis(40)), + round(1, Duration.ofMillis(60))), + ExtractionDiagnostics.GleaningOutcome.COMPLETED) + : new ExtractionDiagnostics( + List.of(round(0, Duration.ofMillis(35))), + ExtractionDiagnostics.GleaningOutcome + .SKIPPED_TOKEN_LIMIT))) + .processNext(); + + GraphRagEventSink.GraphRagEvent glean = capturedStage( + events, GraphRagEventSink.Stage.GLEAN); + assertEquals(2, glean.inputCount(), "both chunks were eligible"); + assertEquals(1, glean.outputCount(), "only one completed a gleaning round"); + assertEquals( + Duration.ofMillis(60), + glean.duration(), + "the first round is EXTRACT's, so only the second round's time is gleaning's cost"); + assertEquals(JOB_ID, glean.operationId(), "the stage belongs to the job that ran it"); + } + + @Test + void reportsNoGleaningStageWhenTheProfileTurnsItOff() { + GraphRagEventSink events = mock(GraphRagEventSink.class); + ClaimedGraphIndex claim = claim( + List.of(chunk(CHUNK_ID, 0, "OrgMemory builds secure retrieval.", null)), + new ExtractionProfile( + "openai", + "gpt-test", + LightRagExtractionPrompt.VERSION, + 40, + 60, + List.of("PRODUCT", "CAPABILITY"), + List.of(), + 0, + 24_000, + 256)); + + processorFor(claim, events, request -> extraction( + request, + ExtractionDiagnostics.notProfiled())) + .processNext(); + + ArgumentCaptor emitted = + ArgumentCaptor.forClass(GraphRagEventSink.GraphRagEvent.class); + verify(events, atLeastOnce()).emit(emitted.capture()); + assertTrue( + emitted.getAllValues().stream().noneMatch(event -> + event.stage() == GraphRagEventSink.Stage.GLEAN), + "a zero-valued series would claim a round that was never configured to run"); + } + @Test void retriesWithoutPublishingWhenTheImmutableEmbeddingRouteDrifts() { GraphIndexingCoordinator coordinator = mock(GraphIndexingCoordinator.class); @@ -394,14 +470,98 @@ private static ClaimedGraphIndex claim() { CHUNK_ID, 0, "OrgMemory builds secure retrieval.", null))); } + /** + * Wires a processor whose only interesting variable is what the extractor reports, so a test + * about gleaning does not have to restate the embedding and publication setup. + */ + private static GraphIndexingProcessor processorFor( + ClaimedGraphIndex claim, + GraphRagEventSink events, + EntityRelationExtractor extractor) { + GraphIndexingCoordinator coordinator = mock(GraphIndexingCoordinator.class); + GraphExtractorFactory extractors = mock(GraphExtractorFactory.class); + EmbeddingModel embeddingModel = mock(EmbeddingModel.class); + AiRouteResolver routes = mock(AiRouteResolver.class); + GraphIndexingProperties properties = properties(); + when(coordinator.claimNext(properties.workerId(), properties.leaseDuration())) + .thenReturn(Optional.of(claim)); + when(routes.resolve(AiWorkload.GRAPH_EXTRACTION)) + .thenReturn(new AiRoute("openai", "gpt-5.6-sol")); + when(routes.resolve(AiWorkload.DOCUMENT_EMBEDDING)) + .thenReturn(new AiRoute("openai", "text-embedding-3-large")); + when(extractors.create(new AiRoute("openai", "gpt-test"))) + .thenReturn(extractor); + when(embeddingModel.embed( + anyList(), isNull(), any(TokenCountBatchingStrategy.class))) + .thenAnswer(invocation -> ((List) invocation.getArgument(0)) + .stream() + .map(ignored -> new float[] {1.0f, 0.0f, 0.0f}) + .toList()); + return new GraphIndexingProcessor( + coordinator, + mock(GraphPublicationCommitter.class), + extractors, + provider(embeddingModel), + routes, + properties, + events); + } + + private static ExtractionResult extraction( + com.orgmemory.graphrag.model.ExtractionRequest request, + ExtractionDiagnostics diagnostics) { + return new ExtractionResult( + request.profile(), + List.of( + new ExtractedEntity( + "source", "OrgMemory", "product", + "Enterprise memory platform", 0.98), + new ExtractedEntity( + "target", "Secure Search", "capability", + "Permission-aware retrieval", 0.97)), + List.of(new ExtractedRelation( + "source", + "target", + "builds", + List.of("security", "retrieval"), + "OrgMemory builds Secure Search", + RelationOrientation.DIRECTED, + 0.96)), + diagnostics); + } + + private static ExtractionRoundMetrics round(int round, Duration elapsed) { + return new ExtractionRoundMetrics(round, 100, 100, 20, elapsed); + } + + private static GraphRagEventSink.GraphRagEvent capturedStage( + GraphRagEventSink events, + GraphRagEventSink.Stage stage) { + ArgumentCaptor emitted = + ArgumentCaptor.forClass(GraphRagEventSink.GraphRagEvent.class); + verify(events, atLeastOnce()).emit(emitted.capture()); + return emitted.getAllValues().stream() + .filter(event -> event.stage() == stage) + .findFirst() + .orElseThrow(() -> new AssertionError("no " + stage + " event was emitted")); + } + private static ClaimedGraphIndex claim(List chunks) { - var graphProcessingProfile = - LightRagGraphProcessingProfiles.current(new ExtractionProfile( + return claim( + chunks, + new ExtractionProfile( "openai", "gpt-test", LightRagExtractionPrompt.VERSION, 40, 60)); + } + + private static ClaimedGraphIndex claim( + List chunks, + ExtractionProfile extractionProfile) { + var graphProcessingProfile = + LightRagGraphProcessingProfiles.current(extractionProfile); var graphProcessingProfileRef = new GraphProcessingProfileRef( UUID.randomUUID(), graphProcessingProfile.canonicalSha256(), diff --git a/docs/increments/active/2026-07-29-observability-pipeline/challenge-generation-telemetry.md b/docs/increments/active/2026-07-29-observability-pipeline/challenge-generation-telemetry.md new file mode 100644 index 000000000..ca662e50e --- /dev/null +++ b/docs/increments/active/2026-07-29-observability-pipeline/challenge-generation-telemetry.md @@ -0,0 +1,84 @@ +# Architecture challenge: where assistant generation telemetry belongs + +You are an independent architecture reviewer. Attack the proposal below rather than validate +it. `CLAUDE.md` requires an independent challenge before a domain-boundary decision is +implemented; this is that challenge. Verify every claim against the code — file paths are +given so you can check rather than trust. + +## The gap + +`GraphRagEventSink.Stage` declares fourteen stages. Production now emits eleven. `GENERATE` +has no producer, so no answer-generation latency, time to first token, or output-side +truncation (`finish_reason=length`) is reported anywhere. A user-visible slow answer is +currently invisible past `ASSEMBLE_CONTEXT`. + +## Why this is not a wiring task + +Generation does not happen inside the GraphRAG runtime, and that is deliberate. + +- `QueryOutputMode` declares `ANSWER` + (`components/graph-rag-core/.../query/QueryOutputMode.java`), so the LightRAG port *can* + generate. +- `GraphRagRetrievalPolicy` pins `CONTEXT` instead + (`core/src/main/java/com/orgmemory/core/knowledge/GraphRagRetrievalPolicy.java:52`). + Generation was moved into the application shell so the shell can re-verify the complete + evidence closure against the canonical ledger before anything reaches a model. +- `AssistantService` (`core/src/main/java/com/orgmemory/core/assistant/AssistantService.java`) + depends on `PermissionAwareKnowledgeSearch` and `ChatModelPort`. +- `PermissionAwareKnowledgeSearch` is engine-neutral and has a second, non-GraphRAG + implementation: `CanonicalHybridKnowledgeSearch`. Which one runs is a configuration switch, + `orgmemory.assistant.retrieval-engine` + (`apps/api/src/main/java/com/orgmemory/api/assistant/AssistantConfiguration.java:46`). +- `GraphRagEvent` requires a non-null `operationId`. Only + `GraphRagKnowledgeRetrievalService` mints one, and it never escapes: + `SecureKnowledgeSearchResult` carries a `requestId` string, evidence, and grounding. + +## Proposal + +Give the assistant turn its own observation surface — meters and a span emitted from the +assistant layer — rather than routing it through `GraphRagEventSink`. Correlate it to +retrieval by trace context rather than by a shared `operationId`. Leave `Stage.GENERATE` +unproduced and remove it from the enum, since the enum would then describe a pipeline +OrgMemory does not run. + +Rationale: the alternative labels canonical-engine turns as GraphRAG stages, which is false, +or threads a GraphRAG operation identifier through an interface that has no such concept, +which inverts the dependency the engine-neutral interface exists to prevent. + +## Strongest counterargument to the proposal + +A second observation surface is a second thing to secure. `GraphRagEventSink` is not merely a +telemetry convenience — it is the enforcement point of the payload boundary. Its compact +constructor structurally rejects free text: fingerprints must match `[0-9a-f]{64}`, +`failureCode` must match `[a-z0-9_]{1,64}`, and no field can carry a prompt. + +Generation is the single highest-risk stage for payload leakage in the entire system: it is +where prompts and completions actually exist. Creating a *new* telemetry path for exactly +that stage, outside the record that makes leakage structurally impossible, is how the +guarantee erodes — not by a decision to weaken it, but by a second path that nobody +remembered to constrain. `ExceptionSanitizingSpanExporter` would still catch exception text, +but nothing would stop a well-meaning `span.setAttribute("completion", …)`. + +The counter-counter is that the boundary can be re-enforced by construction on the new +surface too. But that is a claim about future discipline, and the existing design chose +structure over discipline on purpose. + +## What to decide + +1. Does assistant generation telemetry go through `GraphRagEventSink`, a new payload-free + port with the same structural guarantees, or Micrometer/OpenTelemetry directly? +2. If a new port: what makes its payload boundary structural rather than conventional? +3. Does `Stage.GENERATE` stay in the enum? An enum value nothing can emit is the same class + of defect as an exporter that reports healthy while pushing to nowhere. +4. Does correlation between retrieval and generation come from trace context, or does + `SecureKnowledgeSearchResult` gain an operation identifier? The second is a change to an + engine-neutral interface for one engine's benefit. +5. `finish_reason=length` requires `ChatModelPort` to stop discarding `ChatResponse` + (`integrations/ai-model-gateways/.../SpringAiChatModelAdapter.java:74` calls + `.stream().content()`). Is a richer return type worth it, or should the adapter observe + and report on its own? + +## Scope note + +Time to first token needs no port change — it is measurable by instrumenting the returned +`Flux` — but it still needs a destination, so it is blocked on question 1 alone. 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 fabd7caeb..e01a766c1 100644 --- a/docs/increments/active/2026-07-29-observability-pipeline/plan.md +++ b/docs/increments/active/2026-07-29-observability-pipeline/plan.md @@ -112,13 +112,56 @@ Depends on the composite sink merged in PR #132. adapter calls `.stream().content()`, which discards the `ChatResponse` holding the finish reason. It lands with `GENERATE` below, where the port change is already required. -- [ ] Time to first token on the streaming assistant path. -- [ ] Close the stage gap the LightRAG comparison surfaced. `Stage` declares - fourteen and production emits ten: `PARSE` and `CHUNK` need a sink in the - ingestion pipeline, `GLEAN` needs separating from extraction, and - `GENERATE` needs emitting where retrieval currently stops at - `ASSEMBLE_CONTEXT`. Decide deletion and rebuild separately — it is missing - from the enum entirely and the runbook requires a drill for it. +- [x] Separate `GLEAN` from extraction. The extractor already recorded per-round + metrics and a gleaning outcome and the worker already held them on every + `ExtractedChunk`; nothing published either, so gleaning working and + gleaning silently declined by the token guard were the same picture. The + stage counts eligible chunks against completed rounds and carries the + 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. +- [ ] Deletion and rebuild: missing from the enum entirely, and the runbook + requires a drill for it. Decide separately. +- [ ] Extraction cost. `ExtractionRoundMetrics` already carries + `providerInputTokens` and `providerOutputTokens` per round and nothing + publishes them, so ingestion spend is invisible while retrieval spend is + not. Found while wiring `GLEAN`. The `TokenUsage` record added for context + assembly does not fit — its channels are retrieval's — so this needs its + own shape rather than a forced reuse. + +### `GENERATE` and time to first token — blocked on a boundary decision + +Both need the same answer and neither is a wiring task, so they are recorded +here rather than attempted. + +Generation does not happen inside the GraphRAG runtime. `QueryOutputMode.ANSWER` +exists in the port, but `GraphRagRetrievalPolicy` pins `CONTEXT` +(`core/.../GraphRagRetrievalPolicy.java:52`) and the application shell generates, +so it can re-verify the evidence closure before delivery. `AssistantService` +depends on `PermissionAwareKnowledgeSearch`, which has a second, non-GraphRAG +implementation in `CanonicalHybridKnowledgeSearch`, and on `ChatModelPort`. +`GraphRagEvent` requires a non-null `operationId` that only +`GraphRagKnowledgeRetrievalService` mints and that never leaves it — +`SecureKnowledgeSearchResult` carries a `requestId` string and nothing else. + +So emitting `Stage.GENERATE` from the assistant would either label +canonical-engine turns as GraphRAG stages, or require threading a GraphRAG +operation identifier through an engine-neutral interface that has no such +concept. The alternative — a separate observation surface for the assistant turn +— leaves `Stage.GENERATE` permanently unproduced, which is the gap this item +exists to close. + +`CLAUDE.md` requires an independent architecture challenge before a domain +boundary decision is implemented. `finish_reason=length` needs the same answer +plus a `ChatModelPort` change, because the port streams `Flux` and +`SpringAiChatModelAdapter` calls `.stream().content()`, discarding the +`ChatResponse`. Time to first token needs no port change but still needs a +destination. + +Proposed for challenge, with its strongest counterargument, in +`challenge-generation-telemetry.md`. Cardinality decision, recorded because it is easier to add a tag than to remove one from a series that already exists: organization and operation identifiers, diff --git a/docs/specs/domains/secure-graph-rag.md b/docs/specs/domains/secure-graph-rag.md index 26a7c53ac..963e43dad 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 ten. `PARSE`, `CHUNK`, - `GLEAN` and `GENERATE` have no producer, so no parsing, chunking, gleaning or - answer-generation latency is reported. Deletion and rebuild have no stage. +- `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. +- `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 + guard declining the round is distinguishable from gleaning being configured + off. Its duration is aggregate model time across concurrently gleaned chunks + and is nested inside `EXTRACT`'s wall clock, so stage durations within one job + are not additive. A profile with gleaning disabled emits nothing rather than a + zero. - Two backends ship: an OpenTelemetry span adapter and a Micrometer meter adapter. Neither displaces the other, and each has its own enable property; with both off the producers compose to `NO_OP`. Spans are sampled and show one diff --git a/docs/tests/domains/secure-graph-rag.md b/docs/tests/domains/secure-graph-rag.md index f03ea3708..2a81a1068 100644 --- a/docs/tests/domains/secure-graph-rag.md +++ b/docs/tests/domains/secure-graph-rag.md @@ -67,6 +67,11 @@ 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. +- 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 + is the second round's time alone rather than both rounds'; and that a profile + with gleaning disabled emits no `GLEAN` event at all. - Observability wiring tests prove both backends are contributed together, that either toggle leaves the other in place, and that with both disabled the remaining sinks compose to `NO_OP`.