fix(observability): close the payload bypasses the telemetry boundary did not cover

apps/api/src/main/resources/application.yml

@@ -179,3 +179,19 @@ management:
     web:
       exposure:
         include: health,info,modulith
+
+logging:
+  level:
+    # The provider libraries concatenate the prompt and the response content into
+    # WARN messages of their own, which `spring.ai.chat.observations.log-prompt`
+    # does not reach because it only governs Spring AI observation handlers.
+    # Every such site is guarded by isWarnEnabled(), so pinning these packages
+    # above WARN stops the payload from being built at all.
+    #
+    # This is the safe default, not the enforcement: LOGGING_LEVEL_* environment
+    # variables, system properties and logback configuration all outrank it.
+    # ProviderLoggingBoundaryVerifier checks the resolved level at startup and
+    # refuses to run with these open.
+    # See docs/runbooks/graph-rag-production-hardening.md.
+    org.springframework.ai.openai: ERROR
+    org.springframework.ai.anthropic: ERROR

apps/api/src/test/java/com/orgmemory/api/observability/ProviderLoggingBoundaryTests.java

@@ -0,0 +1,105 @@
+package com.orgmemory.api.observability;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.env.YamlPropertySourceLoader;
+import org.springframework.core.env.PropertySource;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.FileSystemResource;
+
+/**
+ * The OpenAI and Anthropic client libraries build WARN messages that concatenate
+ * the prompt and the response content. {@code spring.ai.chat.observations.log-prompt}
+ * does not reach them because it only governs Spring AI observation handlers.
+ *
+ * <p>Every such call site is guarded by {@code isWarnEnabled()}, so the level decides
+ * whether the payload is built at all. This test covers the shipped default only.
+ * The default is not the enforcement — environment variables and system properties
+ * outrank it — which is what {@code ProviderLoggingBoundaryVerifier} checks at startup
+ * against the resolved level. This test stops the default from drifting.
+ */
+class ProviderLoggingBoundaryTests {
+
+    /** Packages whose own logging concatenates OrgMemory prompt or completion text. */
+    private static final List<String> PAYLOAD_LOGGING_PACKAGES = List.of(
+            "org.springframework.ai.openai",
+            "org.springframework.ai.anthropic");
+
+    /** Levels at or above ERROR; anything else lets a guarded WARN site build its message. */
+    private static final Set<String> LEVELS_ABOVE_WARN = Set.of("ERROR", "FATAL", "OFF");
+
+    @Test
+    void theBaseConfigurationPinsEveryPayloadLoggingPackageAboveWarn() throws IOException {
+        Map<String, Object> properties = load(new ClassPathResource("application.yml"));
+
+        for (String payloadPackage : PAYLOAD_LOGGING_PACKAGES) {
+            // The loader returns origin-tracked char sequences rather than plain strings.
+            Object level = properties.get("logging.level." + payloadPackage);
+            assertEquals(
+                    "ERROR",
+                    String.valueOf(level),
+                    () -> payloadPackage
+                            + " must be pinned to ERROR in application.yml so its guarded WARN"
+                            + " sites cannot concatenate the prompt");
+        }
+    }
+
+    @Test
+    void noProfileLowersAPayloadLoggingPackageBackToWarn() throws IOException {
+        File[] profiles = profileConfigurationFiles();
+        assertTrue(profiles.length > 0, "no application-<profile>.yml was scanned, so this test proves nothing");
+
+        for (File profile : profiles) {
+            Map<String, Object> properties = load(new FileSystemResource(profile));
+
+            properties.forEach((key, value) -> {
+                if (!key.startsWith("logging.level.")) {
+                    return;
+                }
+                String logger = key.substring("logging.level.".length());
+                // Ancestors and descendants both matter: a parent pinned above the packages
+                // relaxes them, and a child such as org.springframework.ai.openai.api overrides
+                // the parent pin for everything under it.
+                if (PAYLOAD_LOGGING_PACKAGES.stream().noneMatch(
+                        p -> p.equals(logger) || p.startsWith(logger + ".") || logger.startsWith(p + "."))) {
+                    return;
+                }
+                assertTrue(
+                        LEVELS_ABOVE_WARN.contains(String.valueOf(value).toUpperCase(Locale.ROOT)),
+                        () -> profile.getName() + " sets " + key + "=" + value
+                                + ", which re-enables the provider WARN sites that log the prompt");
+            });
+        }
+    }
+
+    private static Map<String, Object> load(org.springframework.core.io.Resource resource) throws IOException {
+        List<PropertySource<?>> sources = new YamlPropertySourceLoader().load(resource.getFilename(), resource);
+        if (sources.size() != 1) {
+            fail(resource.getFilename() + " must contain exactly one YAML document, found " + sources.size());
+        }
+        @SuppressWarnings("unchecked")
+        Map<String, Object> properties = (Map<String, Object>) sources.getFirst().getSource();
+        return properties;
+    }
+
+    /**
+     * Profile files live beside {@code application.yml} in this module's resources. Gradle
+     * extracts them to a directory, so listing the parent is enough and keeps the test from
+     * hard-coding a profile list that a new profile would silently escape.
+     */
+    private static File[] profileConfigurationFiles() throws IOException {
+        File resources = new ClassPathResource("application.yml").getFile().getParentFile();
+        File[] profiles = resources.listFiles((directory, name) -> name.startsWith("application-")
+                && (name.endsWith(".yml") || name.endsWith(".yaml")));
+        return profiles == null ? new File[0] : profiles;
+    }
+}

apps/worker/src/main/java/com/orgmemory/worker/graph/GraphIndexingProcessor.java

@@ -79,7 +79,8 @@ class GraphIndexingProcessor {
                 embeddingModels,
                 routes,
                 properties,
-                GraphRagEventSink.composite(eventSinks.orderedStream().toList()));
+                GraphRagEventSink.failureTolerant(
+                        GraphRagEventSink.composite(eventSinks.orderedStream().toList())));
     }
 
     GraphIndexingProcessor(

apps/worker/src/main/resources/application.yml

@@ -114,3 +114,19 @@ management:
   tracing:
     sampling:
       probability: ${ORGMEMORY_TRACING_SAMPLING_PROBABILITY:0.1}
+
+logging:
+  level:
+    # The provider libraries concatenate the prompt and the response content into
+    # WARN messages of their own, which `spring.ai.chat.observations.log-prompt`
+    # does not reach because it only governs Spring AI observation handlers.
+    # Every such site is guarded by isWarnEnabled(), so pinning these packages
+    # above WARN stops the payload from being built at all.
+    #
+    # This is the safe default, not the enforcement: LOGGING_LEVEL_* environment
+    # variables, system properties and logback configuration all outrank it.
+    # ProviderLoggingBoundaryVerifier checks the resolved level at startup and
+    # refuses to run with these open.
+    # See docs/runbooks/graph-rag-production-hardening.md.
+    org.springframework.ai.openai: ERROR
+    org.springframework.ai.anthropic: ERROR

apps/worker/src/test/java/com/orgmemory/worker/observability/ProviderLoggingBoundaryTests.java

@@ -0,0 +1,105 @@
+package com.orgmemory.worker.observability;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.env.YamlPropertySourceLoader;
+import org.springframework.core.env.PropertySource;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.FileSystemResource;
+
+/**
+ * The OpenAI and Anthropic client libraries build WARN messages that concatenate
+ * the prompt and the response content. {@code spring.ai.chat.observations.log-prompt}
+ * does not reach them because it only governs Spring AI observation handlers.
+ *
+ * <p>Every such call site is guarded by {@code isWarnEnabled()}, so the level decides
+ * whether the payload is built at all. The worker runs graph extraction over chunk
+ * content, so its prompts carry evidence text directly. This test covers the shipped
+ * default only; {@code ProviderLoggingBoundaryVerifier} enforces the resolved level at
+ * startup, which is where an environment override is caught.
+ */
+class ProviderLoggingBoundaryTests {
+
+    /** Packages whose own logging concatenates OrgMemory prompt or completion text. */
+    private static final List<String> PAYLOAD_LOGGING_PACKAGES = List.of(
+            "org.springframework.ai.openai",
+            "org.springframework.ai.anthropic");
+
+    /** Levels at or above ERROR; anything else lets a guarded WARN site build its message. */
+    private static final Set<String> LEVELS_ABOVE_WARN = Set.of("ERROR", "FATAL", "OFF");
+
+    @Test
+    void theBaseConfigurationPinsEveryPayloadLoggingPackageAboveWarn() throws IOException {
+        Map<String, Object> properties = load(new ClassPathResource("application.yml"));
+
+        for (String payloadPackage : PAYLOAD_LOGGING_PACKAGES) {
+            // The loader returns origin-tracked char sequences rather than plain strings.
+            Object level = properties.get("logging.level." + payloadPackage);
+            assertEquals(
+                    "ERROR",
+                    String.valueOf(level),
+                    () -> payloadPackage
+                            + " must be pinned to ERROR in application.yml so its guarded WARN"
+                            + " sites cannot concatenate the prompt");
+        }
+    }
+
+    @Test
+    void noProfileLowersAPayloadLoggingPackageBackToWarn() throws IOException {
+        File[] profiles = profileConfigurationFiles();
+        assertTrue(profiles.length > 0, "no application-<profile>.yml was scanned, so this test proves nothing");
+
+        for (File profile : profiles) {
+            Map<String, Object> properties = load(new FileSystemResource(profile));
+
+            properties.forEach((key, value) -> {
+                if (!key.startsWith("logging.level.")) {
+                    return;
+                }
+                String logger = key.substring("logging.level.".length());
+                // Ancestors and descendants both matter: a parent pinned above the packages
+                // relaxes them, and a child such as org.springframework.ai.openai.api overrides
+                // the parent pin for everything under it.
+                if (PAYLOAD_LOGGING_PACKAGES.stream().noneMatch(
+                        p -> p.equals(logger) || p.startsWith(logger + ".") || logger.startsWith(p + "."))) {
+                    return;
+                }
+                assertTrue(
+                        LEVELS_ABOVE_WARN.contains(String.valueOf(value).toUpperCase(Locale.ROOT)),
+                        () -> profile.getName() + " sets " + key + "=" + value
+                                + ", which re-enables the provider WARN sites that log the prompt");
+            });
+        }
+    }
+
+    private static Map<String, Object> load(org.springframework.core.io.Resource resource) throws IOException {
+        List<PropertySource<?>> sources = new YamlPropertySourceLoader().load(resource.getFilename(), resource);
+        if (sources.size() != 1) {
+            fail(resource.getFilename() + " must contain exactly one YAML document, found " + sources.size());
+        }
+        @SuppressWarnings("unchecked")
+        Map<String, Object> properties = (Map<String, Object>) sources.getFirst().getSource();
+        return properties;
+    }
+
+    /**
+     * Profile files live beside {@code application.yml} in this module's resources. Gradle
+     * extracts them to a directory, so listing the parent is enough and keeps the test from
+     * hard-coding a profile list that a new profile would silently escape.
+     */
+    private static File[] profileConfigurationFiles() throws IOException {
+        File resources = new ClassPathResource("application.yml").getFile().getParentFile();
+        File[] profiles = resources.listFiles((directory, name) -> name.startsWith("application-")
+                && (name.endsWith(".yml") || name.endsWith(".yaml")));
+        return profiles == null ? new File[0] : profiles;
+    }
+}

components/graph-rag-core/src/main/java/com/orgmemory/graphrag/observability/FailureTolerantGraphRagEventSink.java

@@ -0,0 +1,96 @@
+package com.orgmemory.graphrag.observability;
+
+import java.lang.System.Logger;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Absorbs a failing telemetry backend without letting the failure disappear.
+ *
+ * <p>Producers already treat emission as non-critical and catch whatever the sink
+ * throws, which is correct — an observability backend must never decide whether a
+ * retrieval or an indexing job succeeds. The cost is that a sink broken since
+ * startup looks exactly like a sink with nothing to report. This wrapper keeps the
+ * absorption and adds the signal that was missing: a running count, the type of the
+ * most recent failure, and one log line the first time each kind of failure appears.
+ *
+ * <p>Only class names are recorded. A telemetry backend's exception message can
+ * quote the request it failed to send, so the message and the stack trace are
+ * treated the same way as any other payload and never leave this class.
+ *
+ * <p>Logging goes through {@link System.Logger} so that
+ * {@code components/graph-rag-core} keeps its property of having no runtime
+ * dependencies; Spring Boot's JUL bridge routes it into the application log.
+ */
+public final class FailureTolerantGraphRagEventSink implements GraphRagEventSink {
+
+    private static final Logger LOGGER =
+            System.getLogger(FailureTolerantGraphRagEventSink.class.getName());
+
+    /**
+     * How many distinct failure types are worth a log line. A backend that produces more
+     * than this is malfunctioning in a way one more line will not clarify.
+     */
+    private static final int REPORTED_FAILURE_TYPE_LIMIT = 10;
+
+    private final GraphRagEventSink delegate;
+    private final AtomicLong swallowedFailures = new AtomicLong();
+    private final AtomicReference<String> lastFailureType = new AtomicReference<>();
+    private final Set<String> reportedFailureTypes = ConcurrentHashMap.newKeySet();
+
+    FailureTolerantGraphRagEventSink(GraphRagEventSink delegate) {
+        this.delegate = Objects.requireNonNull(delegate, "delegate");
+    }
+
+    @Override
+    public void emit(GraphRagEvent event) {
+        try {
+            delegate.emit(event);
+        } catch (RuntimeException failure) {
+            record(failure);
+        }
+    }
+
+    /** How many events this sink has dropped since startup. Never resets. */
+    public long swallowedFailureCount() {
+        return swallowedFailures.get();
+    }
+
+    /** Class name of the most recent failure, or {@code null} while nothing has failed. */
+    public String lastFailureType() {
+        return lastFailureType.get();
+    }
+
+    /** How many log lines this sink has produced. One per distinct failure kind. */
+    int reportedFailureTypeCount() {
+        return reportedFailureTypes.size();
+    }
+
+    private void record(RuntimeException failure) {
+        swallowedFailures.incrementAndGet();
+        String failureType = failure.getClass().getName();
+        lastFailureType.set(failureType);
+        // One line the first time each kind appears. Comparing against the previous kind
+        // alone would flood at event rate as soon as a broken backend alternates, which a
+        // timeout that retries as a connection failure does immediately.
+        if (reportedFailureTypes.size() < REPORTED_FAILURE_TYPE_LIMIT
+                && reportedFailureTypes.add(failureType)) {
+            LOGGER.log(
+                    Logger.Level.WARNING,
+                    "GraphRAG telemetry sink {0} is failing with {1}; events are being dropped."
+                            + " Message and stack trace are withheld because a telemetry failure"
+                            + " can quote the event it could not send. Later failures of this kind"
+                            + " are counted rather than logged.",
+                    delegate.getClass().getName(),
+                    failureType);
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "FailureTolerantGraphRagEventSink{" + delegate + "}";
+    }
+}

components/graph-rag-core/src/main/java/com/orgmemory/graphrag/observability/GraphRagEventSink.java

@@ -60,6 +60,16 @@ static GraphRagEventSink composite(List<GraphRagEventSink> sinks) {
         };
     }
 
+    /**
+     * Wraps a sink so an emission failure is absorbed, counted and reported
+     * instead of being silently discarded at every call site.
+     *
+     * @see FailureTolerantGraphRagEventSink
+     */
+    static FailureTolerantGraphRagEventSink failureTolerant(GraphRagEventSink sink) {
+        return new FailureTolerantGraphRagEventSink(sink);
+    }
+
     record GraphRagEvent(
             UUID operationId,
             UUID organizationId,