feat(agent-observatory): ingest OTLP traces and normalize metrics

Cargo.toml

@@ -171,6 +171,7 @@ serde_path_to_error = "0.1"
 opentelemetry-proto = { version = "0.32", default-features = false, features = [
   "gen-tonic-messages",
   "logs",
+  "metrics",
   "trace",
 ] }
 prost = "0.14"

docs/contracts/agent-observatory.md

@@ -79,11 +79,12 @@ Source kinds and variants are ASCII lower snake case. Event keys are determinist
 The metric point key is SHA-256 hex over the canonical tuple:
 
 ```text
-resource fingerprint, scope, metric name, instrument kind,
-start timestamp, point timestamp, sorted attributes, value, exemplar IDs
+resource fingerprint, scope, metric name, instrument kind, unit,
+aggregation temporality, monotonicity, start timestamp, point timestamp,
+sorted attributes, value, exemplar IDs
 ```
 
-This makes repeated OTLP export idempotent without assuming producer point IDs.
+Unit and applicable aggregation temporality/monotonicity are stream-identifying properties and therefore participate in the key; description does not. `value` is the canonical bounded point payload, including data-point flags. This makes repeated OTLP export idempotent without assuming producer point IDs.
 
 ## 3. Enumerations
 

docs/plans/agent-observatory/proof/PROOF.md

@@ -508,3 +508,32 @@ FIX: adversarial review made nested-array truncation explicit, deterministic att
 PROOF: 6 focused privacy tests and 12 trace tests passed; ordered multi-event/link fixtures preserved exact fields/order, producer and Cortex truncation diagnostics were exact, invalid links/caps rejected safely, path/content opt-ins behaved as configured, and structural-secret negative fixtures proved no planted bearer-token values survived
 REGRESSION: complete OTLP library filter ran 68 tests with 0 failures on the definitive locked harness
 GATE: locked workspace Clippy passed with `-D warnings`; canonical workspace rustfmt, full 500-line production Rust module-size gate, Agent Observatory golden contracts, `git diff --check`, and no Cargo.toml/Cargo.lock drift from AO-042 all passed
+
+## AO-044 Persist trace spans idempotently
+commit/worktree SHA: dce43df4 (checkpoint committed)
+RED: normalized spans had no durable write path, so repeat exports could not prove idempotency or distinguish duplicates from malformed records
+GREEN: added one-transaction `otel_spans` batch persistence with `ON CONFLICT(trace_id, span_id) DO NOTHING`, shared bounded transient-lock retry, write serialization, and explicit accepted/duplicate/rejected accounting
+FIX: direct DB-bypass validation rejects malformed/all-zero IDs, invalid timing, nonexistent run IDs, oversized flattened fields or metadata, and wrong JSON shapes per row without poisoning valid neighbors; the performance cleanup remains intact with no resurrected `OtelSpanRow` scaffold or blanket dead-code allowance
+PROOF: focused locked `db::otlp_traces::tests` passed 5/5 covering empty no-op, repeat-export idempotency, same-batch duplicates, malformed-neighbor isolation, and metadata/flattened-field bounds
+GATE: pre-commit `diff_check`, `env_guard`, 500-line production module-size, and rustfmt hooks passed
+
+## AO-045 Mount functional /v1/traces
+commit/worktree SHA: AO-045 checkpoint (this commit)
+RED: authenticated `/v1/traces` still returned the deferred 404 and had no protobuf decode, bounded request handling, persistence, or OTLP partial-success response
+GREEN: mounted an authenticated protobuf trace endpoint with an 8 MiB route-specific body cap, blocking decode/persistence offloaded through `spawn_blocking`, a 5,000-span request cap, AO-043 privacy-aware normalization, AO-044 idempotent persistence, and encoded `ExportTraceServiceResponse` output
+PARTIAL: malformed individual spans, over-cap spans, configured storage-budget refusal, and direct-storage validation failures are counted as rejected without poisoning valid neighbors; duplicate exports remain successful and do not inflate rejection counts
+STRUCTURE: extracted trace HTTP handling to a focused sidecar so `src/otlp.rs` is 293 lines and `src/otlp/trace_http.rs` is 234 lines, leaving runway for metrics without approaching the 500-line production module gate
+PROOF: definitive locked handler suite passed 13/13 covering valid 200/protobuf, missing and invalid bearer 401, malformed protobuf 400, unsupported media 415, trace 8 MiB and preserved logs 4 MiB body-limit 413 plus Retry-After, invalid-span partial success, 5,000-span cap, storage-budget partial success, and duplicate idempotency
+REGRESSION: definitive locked full OTLP library sweep passed 82/82, including all 5 AO-044 persistence tests plus existing auth/log/normalization/privacy/trace/runtime coverage
+GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and full 500-line production Rust module-size gate passed; real pre-push `cargo clippy --all-targets --all-features --locked -- -D warnings` passed before the AO-045 push
+
+## AO-046 Normalize gauge and sum points
+commit/worktree SHA: AO-046 checkpoint (this commit)
+RED: Cortex did not enable the pinned `opentelemetry-proto` metrics message feature and had no generic normalized metric-point input, number-point converter, or deterministic point key implementation
+GREEN: enabled only the existing pinned metrics feature with no lockfile churn; added privacy-aware integer/double gauge and sum conversion with exact timestamps, gauge start-time semantics, raw temporality integers, monotonicity, provider/session/project identity, canonical resource/scope metadata including entity refs, bounded/sorted point attributes, exemplars, and JSON-safe non-finite double tokens
+IDEMPOTENCY: SHA-256 point keys use fixed-width component framing over canonical resource fingerprint, scope, stream identity, timestamps, sorted attributes, bounded value payload, and sorted exemplar IDs; resource/point/entity-ref input order does not affect the key
+FIX: adversarial review found the original Cortex point-key contract omitted OpenTelemetry stream-identifying unit, aggregation temporality, and monotonicity, which could collapse distinct streams; corrected contract section 2.6 and implementation, preserved data-point flags inside `value_json`, and proved description remains non-identifying
+STRUCTURE: split canonical exemplar/value/key encoding into `metrics_payload.rs`; production metric modules remain comfortably below the 500-line gate
+PROOF: definitive locked focused metric normalization suite passed 8/8, including gauge/sum semantics, deterministic reorder-insensitive keys, stream-identity/flags collision resistance, exemplar validation, non-finite doubles, privacy policy, and fail-closed invalid fields
+REGRESSION: definitive locked full OTLP library sweep passed 90/90 with all prior logs/traces/auth/privacy/runtime/database coverage intact
+GATE: locked production `cargo check` passed without warnings; canonical rustfmt, `git diff --check`, and the full 500-line production Rust module-size gate passed; exact `cargo clippy --all-targets --all-features --locked -- -D warnings` passed after structurally reducing the point-normalizer argument surface and fixing the test config construction lint; the exact four-script Public Identity CI bundle passed after replacing private hostname fixtures with a neutral test host

src/db/otlp_traces.rs

@@ -1,5 +1,22 @@
+//! OTLP trace persistence models and idempotent batch writes.
+
+use anyhow::Result;
+use rusqlite::{Transaction, params};
 use serde::{Deserialize, Serialize};
 
+use super::{DbPool, TRANSIENT_SQLITE_RETRY_DELAYS_MS, is_transient_sqlite_lock, write_lock};
+
+const MAX_METADATA_JSON_BYTES: usize = 256 * 1024;
+const MAX_SPAN_NAME_CHARS: usize = 1024;
+const MAX_TRACE_STATE_CHARS: usize = 512;
+const MAX_STATUS_MESSAGE_CHARS: usize = 4096;
+const MAX_HOSTNAME_CHARS: usize = 255;
+const MAX_SERVICE_CHARS: usize = 512;
+const MAX_SCOPE_CHARS: usize = 512;
+const MAX_TOOL_BYTES: usize = 64;
+const MAX_PROJECT_BYTES: usize = 512;
+const MAX_SESSION_BYTES: usize = 128;
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct OtelSpanInput {
     pub trace_id: String,
@@ -30,3 +47,222 @@ pub struct OtelSpanInput {
     pub received_at: String,
     pub content_scrubbed: bool,
 }
+
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OtelTraceBatchResult {
+    pub accepted: usize,
+    pub duplicates: usize,
+    pub rejected: usize,
+}
+
+impl OtelTraceBatchResult {
+    #[cfg(test)]
+    pub const fn total(self) -> usize {
+        self.accepted + self.duplicates + self.rejected
+    }
+}
+
+pub fn insert_otel_spans_batch(
+    pool: &DbPool,
+    entries: &[OtelSpanInput],
+) -> Result<OtelTraceBatchResult> {
+    let mut attempt = 0usize;
+    loop {
+        match insert_otel_spans_batch_once(pool, entries) {
+            Ok(result) => return Ok(result),
+            Err(error)
+                if is_transient_sqlite_lock(&error)
+                    && attempt < TRANSIENT_SQLITE_RETRY_DELAYS_MS.len() =>
+            {
+                let delay_ms = TRANSIENT_SQLITE_RETRY_DELAYS_MS[attempt];
+                tracing::warn!(
+                    error = %error,
+                    attempt = attempt + 1,
+                    retry_delay_ms = delay_ms,
+                    entry_count = entries.len(),
+                    "Transient SQLite lock during OTLP trace batch insert - retrying"
+                );
+                std::thread::sleep(std::time::Duration::from_millis(delay_ms));
+                attempt += 1;
+            }
+            Err(error) => return Err(error),
+        }
+    }
+}
+
+fn insert_otel_spans_batch_once(
+    pool: &DbPool,
+    entries: &[OtelSpanInput],
+) -> Result<OtelTraceBatchResult> {
+    if entries.is_empty() {
+        return Ok(OtelTraceBatchResult::default());
+    }
+    let mut conn = pool.get()?;
+    let _write_guard = write_lock();
+    let tx = conn.transaction()?;
+    let result = insert_otel_spans_batch_in_tx(&tx, entries)?;
+    tx.commit()?;
+    tracing::debug!(
+        accepted = result.accepted,
+        duplicates = result.duplicates,
+        rejected = result.rejected,
+        "Committed OTLP trace batch transaction"
+    );
+    Ok(result)
+}
+
+fn insert_otel_spans_batch_in_tx(
+    tx: &Transaction<'_>,
+    entries: &[OtelSpanInput],
+) -> Result<OtelTraceBatchResult> {
+    let mut result = OtelTraceBatchResult::default();
+    let mut stmt = tx.prepare_cached(
+        "INSERT INTO otel_spans
+            (trace_id, span_id, parent_span_id, trace_state, flags, span_name, span_kind,
+             start_time_unix_nano, end_time_unix_nano, duration_nano, status_code,
+             status_message, hostname, service_name, service_version, scope_name,
+             scope_version, ai_tool, ai_project, ai_session_id, run_id, resource_json,
+             attributes_json, events_json, links_json, received_at, content_scrubbed)
+         VALUES
+            (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15,
+             ?16, ?17, ?18, ?19, ?20, ?21, ?22, ?23, ?24, ?25, ?26, ?27)
+         ON CONFLICT(trace_id, span_id) DO NOTHING",
+    )?;
+
+    for entry in entries {
+        if !valid_span_input(tx, entry)? {
+            result.rejected += 1;
+            continue;
+        }
+        let changed = stmt.execute(params![
+            entry.trace_id,
+            entry.span_id,
+            entry.parent_span_id,
+            entry.trace_state,
+            entry.flags,
+            entry.span_name,
+            entry.span_kind,
+            entry.start_time_unix_nano,
+            entry.end_time_unix_nano,
+            entry.duration_nano,
+            entry.status_code,
+            entry.status_message,
+            entry.hostname,
+            entry.service_name,
+            entry.service_version,
+            entry.scope_name,
+            entry.scope_version,
+            entry.ai_tool,
+            entry.ai_project,
+            entry.ai_session_id,
+            entry.run_id,
+            entry.resource_json,
+            entry.attributes_json,
+            entry.events_json,
+            entry.links_json,
+            entry.received_at,
+            entry.content_scrubbed,
+        ])?;
+        if changed == 1 {
+            result.accepted += 1;
+        } else {
+            result.duplicates += 1;
+        }
+    }
+    Ok(result)
+}
+
+fn valid_span_input(tx: &Transaction<'_>, entry: &OtelSpanInput) -> Result<bool> {
+    if !valid_hex_id(&entry.trace_id, 32)
+        || !valid_hex_id(&entry.span_id, 16)
+        || entry
+            .parent_span_id
+            .as_deref()
+            .is_some_and(|value| !valid_hex_id(value, 16))
+        || entry.duration_nano < 0
+        || entry.end_time_unix_nano < entry.start_time_unix_nano
+        || entry
+            .end_time_unix_nano
+            .checked_sub(entry.start_time_unix_nano)
+            != Some(entry.duration_nano)
+        || chrono::DateTime::parse_from_rfc3339(&entry.received_at).is_err()
+        || !within_chars(&entry.span_name, MAX_SPAN_NAME_CHARS)
+        || !entry
+            .trace_state
+            .as_deref()
+            .is_none_or(|value| within_chars(value, MAX_TRACE_STATE_CHARS))
+        || !within_chars(&entry.hostname, MAX_HOSTNAME_CHARS)
+        || !optional_chars(&entry.service_name, MAX_SERVICE_CHARS)
+        || !optional_chars(&entry.service_version, MAX_SERVICE_CHARS)
+        || !optional_chars(&entry.scope_name, MAX_SCOPE_CHARS)
+        || !optional_chars(&entry.scope_version, MAX_SCOPE_CHARS)
+        || !optional_chars(&entry.status_message, MAX_STATUS_MESSAGE_CHARS)
+        || !optional_bytes(&entry.ai_tool, MAX_TOOL_BYTES)
+        || !optional_bytes(&entry.ai_project, MAX_PROJECT_BYTES)
+        || !optional_bytes(&entry.ai_session_id, MAX_SESSION_BYTES)
+        || !json_shape(&entry.resource_json, JsonShape::Object)
+        || !json_shape(&entry.attributes_json, JsonShape::Object)
+        || !json_shape(&entry.events_json, JsonShape::Array)
+        || !json_shape(&entry.links_json, JsonShape::Array)
+    {
+        return Ok(false);
+    }
+    if let Some(run_id) = entry.run_id {
+        if run_id <= 0 {
+            return Ok(false);
+        }
+        let exists: bool = tx.query_row(
+            "SELECT EXISTS(SELECT 1 FROM agent_runs WHERE id = ?1)",
+            [run_id],
+            |row| row.get(0),
+        )?;
+        if !exists {
+            return Ok(false);
+        }
+    }
+    Ok(true)
+}
+
+fn valid_hex_id(value: &str, expected_chars: usize) -> bool {
+    value.len() == expected_chars
+        && value.bytes().all(|byte| byte.is_ascii_hexdigit())
+        && value.bytes().any(|byte| byte != b'0')
+}
+
+fn within_chars(value: &str, maximum: usize) -> bool {
+    value.chars().count() <= maximum
+}
+
+fn optional_chars(value: &Option<String>, maximum: usize) -> bool {
+    value
+        .as_deref()
+        .is_none_or(|value| within_chars(value, maximum))
+}
+
+fn optional_bytes(value: &Option<String>, maximum: usize) -> bool {
+    value.as_ref().is_none_or(|value| value.len() <= maximum)
+}
+
+#[derive(Clone, Copy)]
+enum JsonShape {
+    Object,
+    Array,
+}
+
+fn json_shape(raw: &str, shape: JsonShape) -> bool {
+    if raw.len() > MAX_METADATA_JSON_BYTES {
+        return false;
+    }
+    let Ok(value) = serde_json::from_str::<serde_json::Value>(raw) else {
+        return false;
+    };
+    matches!(
+        (shape, value),
+        (JsonShape::Object, serde_json::Value::Object(_))
+            | (JsonShape::Array, serde_json::Value::Array(_))
+    )
+}
+
+#[cfg(test)]
+#[path = "otlp_traces_tests.rs"]
+mod tests;