feat(graph-rag): report gleaning as its own extraction stage

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> T observed(
         }
     }
 
+    /**
+     * Reports the second extraction round separately from the first.
+     *
+     * <p>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.
+     *
+     * <p>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.
+     *
+     * <p>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<ExtractedChunk> 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,

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<GraphRagEventSink.GraphRagEvent> 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<GraphRagEventSink.GraphRagEvent> 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<Document>) 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<GraphRagEventSink.GraphRagEvent> 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<GraphIndexChunk> chunks) {
-        var graphProcessingProfile =
-                LightRagGraphProcessingProfiles.current(new ExtractionProfile(
+        return claim(
+                chunks,
+                new ExtractionProfile(
                         "openai",
                         "gpt-test",
                         LightRagExtractionPrompt.VERSION,
                         40,
                         60));
+    }
+
+    private static ClaimedGraphIndex claim(
+            List<GraphIndexChunk> chunks,
+            ExtractionProfile extractionProfile) {
+        var graphProcessingProfile =
+                LightRagGraphProcessingProfiles.current(extractionProfile);
         var graphProcessingProfileRef = new GraphProcessingProfileRef(
                 UUID.randomUUID(),
                 graphProcessingProfile.canonicalSha256(),

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.