fix(sensor): acknowledge OTLP delivery independently of local capture

Sensor/README.md

@@ -354,11 +354,43 @@ no redaction or field projection, so prompts, responses, tool arguments, tool
 results, usernames, hostnames, and local paths can be transmitted. Any
 normalization already performed by a source parser still applies.
 
-System-configuration records are sent as `adr.system.configuration` logs. Runs
-are not checkpointed specifically for OTLP: repeated runs can resend the same
-records, and consumers can use `adr.event.uuid` to deduplicate them.
+System-configuration records are sent as `adr.system.configuration` logs on each
+run. Sensor health logs are also sent on every run, even when all session snapshots
+are already acknowledged. With `--save-sessions`, successful session delivery is tracked independently
+of local session files. A failed export is retried on the next run, even when the
+local JSON already exists. A session is skipped only when its complete normalized
+payload was successfully exported to the same destination configuration. Changes
+to tool results, destination settings, or configured authentication headers cause
+a resend. The checkpoint also accounts for effective OTLP environment headers and
+mTLS client certificate/key paths. It does not read credential files: after
+changing certificate or key contents in place, remove the destination's checkpoint
+to resend sessions. Dynamic HTTP credential-provider plugins
+(`OTEL_PYTHON_EXPORTER_OTLP_HTTP_CREDENTIAL_PROVIDER` and its `LOGS` variant)
+are unsupported and cause an explicit error; use configured headers or mTLS.
+
+Delivery checkpoints are hidden `.adr-otel-delivery.<hash>.json` files in the
+session output directory. They contain only hashes, including a destination hash
+that accounts for authentication headers; they do not store raw URLs, credentials,
+session identifiers, or payloads. Missing, unreadable, or corrupt checkpoints cause
+sessions to be retried. The checkpoint is replaced atomically only after flush and
+shutdown succeed; a checkpoint write failure exits with an error. `--no-save`
+disables checkpoint reads and writes. Without `--save-sessions`, every run exports
+all collected sessions.
+
+The one-shot Sensor process drains bounded batches and reconciles submitted and
+successfully exported counts before reporting success, so a full SDK queue cannot
+silently drop records. HTTP success is also checked for an OTLP acknowledgement:
+partial rejection or a malformed response fails delivery and leaves the affected
+run unacknowledged. Resolve persistent collector rejection before rerunning: OTLP
+does not identify individual rejected records, so retrying can resend accepted
+records too. Export or checkpoint failures exit with a nonzero status.
+Delivery is at least once: a collector may receive data before a timeout, process
+interruption, or checkpoint write failure, so retries can duplicate records.
+Checkpointing only covers sessions that are collected again on a later run; it is
+not a persistent payload queue. Consumers can use `adr.event.uuid` and a full
+payload digest to identify repeated snapshots, since a session UUID alone does not
+necessarily change when tool results change.
 
-The one-shot Sensor process flushes and shuts down the exporter before exiting.
 Use an OpenTelemetry Collector when vendor-specific routing, transformation,
 retry, or persistent queuing is needed.
 

Sensor/adr_sensor/cli.py

@@ -26,6 +26,7 @@
 from . import __version__
 from .diagnostics import health_record, write_health_records
 from .exporters import OpenTelemetryConfigError, load_opentelemetry_config
+from .exporters.delivery_checkpoint import DeliveryCheckpoint, DeliveryCheckpointError
 from .exporters.opentelemetry import OpenTelemetryExportError, OpenTelemetryLogExporter
 from .observer import AgentObserver
 
@@ -150,7 +151,21 @@ def main():
         stage = "parse"
         entries, system_config_data = observer.ingest_all(args.source)
 
+        # A local file is not an OTLP acknowledgement. Keep remote candidates
+        # independent of local incremental filtering, including after a failed run.
+        otel_entries = entries
+        delivery_checkpoint = None
+        if otel_config is not None and args.save_sessions and not args.no_save:
+            stage = "export"
+            checkpoint_dir = args.output_dir if args.output_dir is not None else observer._get_default_session_dir()
+            delivery_checkpoint = DeliveryCheckpoint(checkpoint_dir, otel_config)
+            otel_entries = delivery_checkpoint.pending_entries(entries)
+            if delivery_checkpoint.load_failed:
+                observer.record_failure("export", "checkpoint_read_error")
+                print("OpenTelemetry delivery checkpoint unreadable or invalid; retrying sessions.", file=sys.stderr)
+
         # Apply incremental filtering
+        stage = "save"
         if args.save_sessions and entries:
             print("\nSession-based incremental mode: Checking existing session files...")
             original_count = len(entries)
@@ -186,10 +201,12 @@ def main():
             stage = "export"
             otel_exporter = OpenTelemetryLogExporter(otel_config, service_version=get_version())
             try:
-                exported_count = otel_exporter.export(entries, system_config_data)
+                exported_count = otel_exporter.export(otel_entries, system_config_data)
                 otel_exporter.export_diagnostics(observer.get_diagnostic_records())
             finally:
                 otel_exporter.shutdown()
+            if delivery_checkpoint is not None:
+                delivery_checkpoint.commit()
             print(f"\nOpenTelemetry session/configuration logs sent: {exported_count}")
 
         success = observer.has_errors is not True
@@ -207,6 +224,13 @@ def main():
         print(f"OpenTelemetry export failed: {exc}", file=sys.stderr)
         raise SystemExit(1)
 
+    except DeliveryCheckpointError as exc:
+        success = False
+        if observer is not None:
+            observer.record_failure("export", "checkpoint_write_error")
+        print(f"OpenTelemetry checkpoint failed: {exc}", file=sys.stderr)
+        raise SystemExit(1)
+
     except Exception:
         success = False
         if observer is not None:

Sensor/adr_sensor/diagnostics.py

@@ -15,7 +15,9 @@
     {"sensor", "claude", "claude_desktop", "cursor", "cline", "codex", "copilot", "dsh", "gemini", "opencode", "warp"}
 )
 DIAGNOSTIC_STAGES = frozenset({"parse", "save", "save_session", "export", "startup"})
-OPERATIONAL_REASONS = frozenset({"parser_error", "write_error", "export_error", "startup_error"})
+OPERATIONAL_REASONS = frozenset(
+    {"parser_error", "write_error", "export_error", "startup_error", "checkpoint_read_error", "checkpoint_write_error"}
+)
 COUNT_FIELDS = frozenset({"events_returned", "events_emitted", "events_filtered", "attempted", "succeeded", "failed"})
 MAX_LOG_BYTES = 1024 * 1024
 LOG_BACKUP_COUNT = 2

Sensor/adr_sensor/exporters/delivery_checkpoint.py

@@ -0,0 +1,120 @@
+"""Hash-only acknowledgements for incremental OTLP session delivery."""
+
+import hashlib
+import json
+import os
+import re
+import tempfile
+from dataclasses import asdict
+from pathlib import Path
+from typing import Dict, List
+
+from ..schemas.agent_event_schema import AgentEvent
+from .config import OpenTelemetryConfig
+from .opentelemetry import SCHEMA_VERSION, _validate_credential_provider
+
+_DIGEST = re.compile(r"[0-9a-f]{64}\Z")
+
+
+class DeliveryCheckpointError(RuntimeError):
+    """Raised when a successful export cannot be durably checkpointed."""
+
+
+def _fingerprint(value: object) -> str:
+    serialized = json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
+    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
+
+
+class DeliveryCheckpoint:
+    """Remember delivered session snapshots separately from their local JSON files.
+
+    Select pending entries before export, then call commit only after flush and
+    shutdown succeed. Losing or corrupting this cache causes retries, never skips.
+    """
+
+    def __init__(self, output_dir: Path, config: OpenTelemetryConfig):
+        _validate_credential_provider()
+        destination = {
+            "config": asdict(config),
+            "schema_version": SCHEMA_VERSION,
+            "checkpoint_version": 1,
+        }
+        if not config.headers:
+            # The HTTP exporter falls back to these variables for empty headers.
+            destination["environment_headers"] = os.environ.get(
+                "OTEL_EXPORTER_OTLP_LOGS_HEADERS", os.environ.get("OTEL_EXPORTER_OTLP_HEADERS", "")
+            )
+        for setting in ("CLIENT_CERTIFICATE", "CLIENT_KEY"):
+            destination[setting] = os.environ.get(
+                f"OTEL_EXPORTER_OTLP_LOGS_{setting}", os.environ.get(f"OTEL_EXPORTER_OTLP_{setting}", "")
+            )
+        self.path = Path(output_dir) / f".adr-otel-delivery.{_fingerprint(destination)}.json"
+        self.load_failed = False
+        self._delivered = self._load()
+        self._pending: Dict[str, str] = {}
+
+    def _load(self) -> Dict[str, str]:
+        try:
+            with self.path.open(encoding="utf-8") as checkpoint_file:
+                data = json.load(checkpoint_file)
+            if not isinstance(data, dict) or not all(
+                isinstance(key, str) and _DIGEST.fullmatch(key) and isinstance(value, str) and _DIGEST.fullmatch(value)
+                for key, value in data.items()
+            ):
+                raise ValueError("invalid checkpoint fingerprints")
+            return data
+        except FileNotFoundError:
+            return {}
+        except (OSError, ValueError):
+            self.load_failed = True
+            return {}
+
+    def pending_entries(self, entries: List[AgentEvent]) -> List[AgentEvent]:
+        """Select snapshots not acknowledged for this destination, including results."""
+        self._pending = {}
+        pending_entries = []
+        for entry in entries:
+            identity = _fingerprint([entry.source, entry.session_id, entry.hostname, entry.username])
+            payload = _fingerprint(entry.get_non_null_fields())
+            if self._delivered.get(identity) != payload:
+                pending_entries.append(entry)
+                self._pending[identity] = payload
+        return pending_entries
+
+    def commit(self) -> None:
+        """Atomically persist only the snapshots selected for a successful export."""
+        if not self._pending:
+            return
+        temporary_path = None
+        try:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            # Merge recent acknowledgements; concurrent writers may cause extra
+            # retries, but can never acknowledge a payload they have not exported.
+            delivered = self._load()
+            delivered.update(self._pending)
+            fd, temporary_name = tempfile.mkstemp(prefix=f"{self.path.name}.", suffix=".tmp", dir=self.path.parent)
+            temporary_path = Path(temporary_name)
+            with os.fdopen(fd, "w", encoding="utf-8") as checkpoint_file:
+                json.dump(delivered, checkpoint_file, sort_keys=True, separators=(",", ":"))
+                checkpoint_file.flush()
+                os.fsync(checkpoint_file.fileno())
+            os.replace(temporary_path, self.path)
+            temporary_path = None
+            if os.name != "nt":
+                directory_fd = os.open(self.path.parent, os.O_RDONLY)
+                try:
+                    os.fsync(directory_fd)
+                finally:
+                    os.close(directory_fd)
+            self._delivered = delivered
+            self._pending = {}
+        except OSError as exc:
+            raise DeliveryCheckpointError(
+                "could not save the OpenTelemetry delivery checkpoint; a later run may resend delivered sessions"
+            ) from exc
+        finally:
+            if temporary_path is not None:
+                try:
+                    temporary_path.unlink()
+                except OSError:
+                    pass