[sdk] Native support for LangChain/LangGraph, CrewAI, LlamaIndex and Pydantic AI, on a real identity layer

.github/workflows/ci.yml

@@ -3,8 +3,14 @@ name: CI
 on:
   push:
     branches: [main]
+  # `main` AND the long-lived branches other PRs stack onto. A pull request is
+  # tested against the branch it will actually merge into, and a stacked PR
+  # targeting anything but `main` matched nothing here — so #730, thirty commits
+  # of SDK work, ran no unit tests, no build and no lint at all. The only signal
+  # it produced was the daemon cross-compile, and only because it touched
+  # `crates/`. A PR that cannot go red is not a reviewed PR.
   pull_request:
-    branches: [main]
+    branches: [main, feat/fp-cli]
 
 concurrency:
   group: ci-${{ github.ref }}
@@ -263,6 +269,9 @@ jobs:
   # claiming >=3.10 and testing only one of them is how a 3.10 user finds the break.
   fp-cli:
     runs-on: ubuntu-latest
+    # a uv sync plus pytest across two interpreters; the bound exists so a stalled
+    # package mirror cannot hold a release for six hours (#726).
+    timeout-minutes: 10
     defaults:
       run:
         working-directory: fp-cli
@@ -334,6 +343,9 @@ jobs:
   # requires-python advertises is tested.
   failproofai-sdk:
     runs-on: ubuntu-latest
+    # the same, across five — every version requires-python advertises; the bound exists so a stalled
+    # package mirror cannot hold a release for six hours (#726).
+    timeout-minutes: 10
     defaults:
       run:
         working-directory: sdk/python

crates/failproofaid/src/main.rs

@@ -734,6 +734,9 @@ fn collector_tasks() -> Vec<fpai_collect::TaskSpec> {
     // One `Delivery` shared by both tasks, so they share an upload semaphore
     // and an in-flight set. Separate ones would let the watcher and a
     // concurrent sweep POST the same batch twice.
+    // Grabbed before the uploader moves into `Delivery`. The counters live on
+    // the `Uploader` itself so a supervised task restart never rewinds them.
+    let upload_metrics = uploader.metrics();
     let delivery = std::sync::Arc::new(fpai_collect::Delivery::new(uploader));
 
     let watch_delivery = delivery.clone();
@@ -750,6 +753,12 @@ fn collector_tasks() -> Vec<fpai_collect::TaskSpec> {
     // daemon", where a stale file makes a stopped daemon look like a running
     // one whose sources all went quiet.
     let health = std::sync::Arc::new(fpai_collect::Health::new());
+    // Before `install`, so the first snapshot already carries delivery. The
+    // source map alone cannot say whether anything is ARRIVING — a source's job
+    // ends at the spool — and the SDK's batches have no source entry at all, so
+    // a machine shipping only SDK events reported an empty, healthy-looking file
+    // whether ingest was storing every event or discarding all of them.
+    health.attach_delivery(upload_metrics);
     fpai_collect::health::install(health.clone());
     let health_file = fpai_collect::health_path(&home);
     tasks.push(fpai_collect::TaskSpec::new("health", move |sd| {

crates/failproofaid/tests/collector_reload_e2e.rs

@@ -256,22 +256,57 @@ fn disabling_collection_stops_it_and_re_enabling_starts_it_again() {
     // `--disconnect` used to require a restart to take effect, so a machine that
     // had left its organisation went on shipping. The reverse matters just as
     // much: re-enabling must not need one either, or the fix is half a fix.
+    //
+    // The lever is the CREDENTIAL, because that is what `--disconnect` actually
+    // removes (`clearIngestCredential` in cloud-enrollment-cli.ts). This used to
+    // toggle `collector.hooks` as a stand-in, which no longer disables anything:
+    // that flag gates the daemon's own hook-activity source and deliberately
+    // does NOT gate delivery, since the spool also carries batches the
+    // `failproofai-sdk` wrote from the user's own process. Toggling the real
+    // thing is a stronger test of the scenario this exists for, and the
+    // companion test below pins the behaviour that replaced the old lever.
     let home = unique_home("toggle");
     make_home(&home, "a-stable-key");
     let daemon = spawn_daemon(&home);
     daemon.wait_for("collector enabled", 1, Duration::from_secs(20));
 
+    let creds = home.join("credentials.json");
+    let on = std::fs::read_to_string(&creds).unwrap();
+    std::fs::remove_file(&creds).unwrap();
+    daemon.wait_for("no longer enabled", 1, Duration::from_secs(20));
+
+    std::fs::write(&creds, &on).unwrap();
+    daemon.wait_for("collector enabled", 2, Duration::from_secs(20));
+    let _ = std::fs::remove_dir_all(&home);
+}
+
+#[test]
+fn turning_both_capture_sources_off_leaves_delivery_running() {
+    // The counterweight to the change above, and the bug it came from: with
+    // `sessions` and `hooks` both false the daemon used to start NOTHING — no
+    // spool watcher, no sweeper, no log line — so every batch the SDK wrote
+    // sat on disk forever, with no error on either side and an unread spool
+    // indistinguishable from an idle one.
+    //
+    // Those two settings gate the daemon's own capture sources, and each is
+    // checked again where its source is registered, so leaving them off still
+    // starts neither. What they must not gate is delivery.
+    let home = unique_home("sources-off");
+    make_home(&home, "a-stable-key");
     let cfg = home.join("config.json");
     let on = std::fs::read_to_string(&cfg).unwrap();
-    // JSON now, so the edit is on the key/value pair, not a TOML line. A
-    // string replace that silently matches nothing writes the file back
-    // unchanged and the test then waits 20s for a reload that never had a
-    // reason to happen — which is exactly how this broke.
     std::fs::write(&cfg, on.replace(r#""hooks":true"#, r#""hooks":false"#)).unwrap();
-    daemon.wait_for("no longer enabled", 1, Duration::from_secs(20));
 
-    std::fs::write(&cfg, &on).unwrap();
-    daemon.wait_for("collector enabled", 2, Duration::from_secs(20));
+    let daemon = spawn_daemon(&home);
+    daemon.wait_for("collector enabled", 1, Duration::from_secs(20));
+    // The watcher is the thing that ships an SDK batch; without it this daemon
+    // is a process that reports healthy and delivers nothing.
+    daemon.wait_for("spool watcher started", 1, Duration::from_secs(20));
+    assert!(
+        !daemon.stderr().contains("hook-activity source started"),
+        "hooks=false must still switch the daemon's own hook source off:\n{}",
+        daemon.stderr()
+    );
     let _ = std::fs::remove_dir_all(&home);
 }
 

crates/fpai-collect/src/config.rs

@@ -261,11 +261,25 @@ pub struct CollectorConfig {
 }
 
 impl CollectorConfig {
-    /// True when there is a usable credential AND at least one stream enabled.
-    /// This is what `collector_tasks()` keys off, so an unconfigured machine
-    /// starts no thread and no runtime.
+    /// True when there is a usable credential. This is what `collector_tasks()`
+    /// keys off, so an unconfigured machine starts no thread and no runtime.
+    ///
+    /// Deliberately NOT `&& (sessions || hooks)`. Those two gate the daemon's
+    /// OWN capture sources — CLI session transcripts and hook activity — and
+    /// each is checked again where its source is registered, so leaving them
+    /// off still starts neither. What they must not gate is DELIVERY, because
+    /// the spool also carries batches this daemon did not produce: the
+    /// `failproofai-sdk` writes its own events into `custom-agents/events/`,
+    /// and the spool watcher is the only thing that ships them.
+    ///
+    /// While they did gate it, `collector.hooks = false` — a documented
+    /// privacy choice, and the only one available to somebody who wants their
+    /// instrumented agents shipped and nothing else — silently disabled the
+    /// SDK too: no task started, no line logged, and batches accumulated in
+    /// the spool forever. An unread spool is indistinguishable from an idle
+    /// one, which is the exact failure this project exists to remove.
     pub fn is_enabled(&self) -> bool {
-        self.ingest.is_some() && (self.settings.sessions || self.settings.hooks)
+        self.ingest.is_some()
     }
 }
 

crates/fpai-collect/src/health.rs

@@ -55,13 +55,49 @@ pub struct SourceHealth {
     pub errors: u64,
 }
 
+/// What delivery reports about itself, across every source AND the SDK spool.
+///
+/// The source map above cannot answer "is anything actually arriving", because
+/// a source's job ends when it writes a batch to the spool. Everything after
+/// that — the POST, the server's verdict, the parking of what would not go — is
+/// invisible to it, and the SDK's own batches have no source entry at all: the
+/// `failproofai-sdk` writes them straight into the spool from the user's
+/// process, so a machine shipping nothing but SDK events reports a perfectly
+/// healthy, entirely empty `sources` map.
+///
+/// `skipped` is the one worth staring at. Ingest answers `200` with
+/// `{"accepted":N,"skipped":M}` and the daemon deletes the batch either way, so
+/// a systematically malformed field — an `environment` containing a comma, say
+/// — discards every event on the machine while every layer reports success.
+/// Before this, the only trace was a line in the daemon's log.
+#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+pub struct DeliveryHealth {
+    /// Events the server said it stored, since this daemon started.
+    pub accepted: u64,
+    /// Events the server refused as malformed. Non-zero means data loss that
+    /// nothing else on the machine will tell you about.
+    pub skipped: u64,
+    /// Batches answered `200` while storing NONE of their events — the shape a
+    /// systematic problem takes, as opposed to one bad line.
+    pub batches_fully_skipped: u64,
+    /// Unix seconds of the last upload the server accepted. Zero means not one
+    /// has succeeded since startup, which on a machine that is producing events
+    /// is the loudest thing in this file.
+    pub last_ok_ts: u64,
+}
+
 /// The whole record, as written.
 #[derive(Debug, Default, Deserialize, Serialize)]
 pub struct HealthFile {
     /// When this snapshot was written, so a reader can tell a stale file from a
     /// current one — a daemon that died leaves its last record behind.
     pub ts: u64,
     pub sources: BTreeMap<String, SourceHealth>,
+    /// Absent when this daemon has no uploader, which is every daemon with no
+    /// credential. Skipped rather than zeroed: all-zero counters and "delivery
+    /// is not configured" are different facts and must not render the same.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub delivery: Option<DeliveryHealth>,
 }
 
 /// Shared, cheap-to-update health state.
@@ -73,6 +109,9 @@ pub struct HealthFile {
 pub struct Health {
     sources: Mutex<BTreeMap<String, SourceHealth>>,
     writes: AtomicU64,
+    /// Set once at startup when there is an uploader. Read, never written, so
+    /// the counters stay owned by the `Uploader` that survives a task restart.
+    delivery: Mutex<Option<std::sync::Arc<crate::UploadMetrics>>>,
 }
 
 impl Health {
@@ -109,12 +148,28 @@ impl Health {
         entry.last_error = Some(truncate(error, MAX_ERROR_LEN));
     }
 
+    /// Report delivery counters alongside the sources. Called once, at startup.
+    pub fn attach_delivery(&self, metrics: std::sync::Arc<crate::UploadMetrics>) {
+        if let Ok(mut slot) = self.delivery.lock() {
+            *slot = Some(metrics);
+        }
+    }
+
     /// Snapshot for writing.
     pub fn snapshot(&self) -> HealthFile {
         let sources = self.sources.lock().map(|m| m.clone()).unwrap_or_default();
+        let delivery = self.delivery.lock().ok().and_then(|slot| {
+            slot.as_ref().map(|m| DeliveryHealth {
+                accepted: m.accepted_total.load(Ordering::Relaxed),
+                skipped: m.skipped_total.load(Ordering::Relaxed),
+                batches_fully_skipped: m.batches_fully_skipped.load(Ordering::Relaxed),
+                last_ok_ts: m.last_ok_ts.load(Ordering::Relaxed),
+            })
+        });
         HealthFile {
             ts: now_secs(),
             sources,
+            delivery,
         }
     }
 

crates/fpai-collect/src/lib.rs

@@ -31,7 +31,7 @@ pub use config::{
 };
 pub use delivery::Delivery;
 pub use extra_paths::{ExtraPath, Resolved as ResolvedExtraPaths};
-pub use health::{Health, HealthFile, SourceHealth, health_path};
+pub use health::{DeliveryHealth, Health, HealthFile, SourceHealth, health_path};
 pub use spool::SpoolWriter;
 pub use supervisor::{
     CollectorHandle, DEFAULT_FLUSH_BUDGET, Shutdown, SupervisorMetrics, TaskError, TaskSpec,

crates/fpai-collect/tests/config.rs

@@ -79,6 +79,39 @@ fn a_key_alone_does_not_enable_session_collection() {
     fs::remove_dir_all(&home).ok();
 }
 
+#[test]
+fn a_key_with_both_sources_off_still_delivers_the_sdk_spool() {
+    // `collector.hooks = false` is a documented privacy choice. It must switch
+    // off the daemon's own hook-activity source and NOTHING else: the spool it
+    // watches also holds batches written by `failproofai-sdk` from the user's
+    // own instrumented agents, and the watcher is the only thing that ships
+    // them. While this returned false for that config the daemon started no
+    // task and logged no line, and those batches piled up forever.
+    let home = tmp_home("bothoff");
+    config::write_ingest(
+        &home,
+        &Ingest {
+            url: DEFAULT_INGEST_URL.into(),
+            key: "k".into(),
+        },
+    )
+    .unwrap();
+    fs::write(
+        home.join("config.json"),
+        r#"{"collector":{"sessions":false,"hooks":false}}"#,
+    )
+    .unwrap();
+
+    let cfg = without_env_overrides(|| config::load(&home).unwrap());
+    assert!(!cfg.settings.sessions);
+    assert!(!cfg.settings.hooks);
+    assert!(
+        cfg.is_enabled(),
+        "delivery must run for the SDK spool even with both capture sources off"
+    );
+    fs::remove_dir_all(&home).ok();
+}
+
 #[test]
 fn the_credential_file_is_written_owner_only() {
     #[cfg(unix)]

crates/fpai-collect/tests/health_delivery.rs

@@ -0,0 +1,91 @@
+//! The health file must be able to say whether anything is ARRIVING.
+//!
+//! Its source map cannot. A source's job ends when it writes a batch into the
+//! spool — the POST, the server's verdict and the parking of what would not go
+//! are all after that — and the SDK's batches have no source entry at all,
+//! because `failproofai-sdk` writes them into the spool from the user's own
+//! process. So a machine shipping nothing but SDK events produced a file with
+//! an empty, perfectly healthy `sources` map whether ingest was storing every
+//! event or discarding all of them.
+
+use std::sync::Arc;
+use std::sync::atomic::Ordering;
+
+use fpai_collect::{Health, HealthFile, UploadMetrics};
+
+#[test]
+fn delivery_is_absent_until_an_uploader_attaches() {
+    // All-zero counters and "this daemon has no credential, so nothing is being
+    // delivered at all" are different facts. Rendering them the same would make
+    // an unconfigured machine look like a broken one, and vice versa.
+    let health = Health::new();
+    let json = serde_json::to_string(&health.snapshot()).unwrap();
+    assert!(
+        !json.contains("delivery"),
+        "an unconfigured daemon must omit the section, not zero it: {json}"
+    );
+}
+
+#[test]
+fn attached_delivery_counters_reach_the_snapshot() {
+    let health = Health::new();
+    let metrics = Arc::new(UploadMetrics::default());
+    health.attach_delivery(metrics.clone());
+
+    metrics.accepted_total.fetch_add(120, Ordering::Relaxed);
+    metrics.skipped_total.fetch_add(7, Ordering::Relaxed);
+    metrics
+        .batches_fully_skipped
+        .fetch_add(1, Ordering::Relaxed);
+    metrics.last_ok_ts.store(1_787_216_518, Ordering::Relaxed);
+
+    let snap = health.snapshot();
+    let delivery = snap
+        .delivery
+        .expect("delivery must be reported once attached");
+    assert_eq!(delivery.accepted, 120);
+    assert_eq!(
+        delivery.skipped, 7,
+        "skipped is the counter that means data loss"
+    );
+    assert_eq!(delivery.batches_fully_skipped, 1);
+    assert_eq!(delivery.last_ok_ts, 1_787_216_518);
+}
+
+#[test]
+fn the_counters_are_read_live_rather_than_copied_at_attach() {
+    // The `Uploader` owns them and outlives any supervised task restart, so the
+    // health writer must read through to it. Snapshotting the values at attach
+    // time would freeze the file at "nothing has happened yet" forever — which
+    // reads exactly like a healthy idle machine.
+    let health = Health::new();
+    let metrics = Arc::new(UploadMetrics::default());
+    health.attach_delivery(metrics.clone());
+    assert_eq!(health.snapshot().delivery.unwrap().accepted, 0);
+
+    metrics.accepted_total.fetch_add(5, Ordering::Relaxed);
+    assert_eq!(health.snapshot().delivery.unwrap().accepted, 5);
+}
+
+#[test]
+fn a_written_file_round_trips_through_the_published_type() {
+    // `HealthFile` is what a reader outside this crate deserializes; an added
+    // field that only serializes one way would be invisible to them.
+    let dir = std::env::temp_dir().join(format!("fpai-health-{}", std::process::id()));
+    let path = dir.join("collector-health.json");
+    let health = Health::new();
+    let metrics = Arc::new(UploadMetrics::default());
+    metrics.skipped_total.fetch_add(3, Ordering::Relaxed);
+    health.attach_delivery(metrics);
+    health.write(&path).unwrap();
+
+    let parsed: HealthFile =
+        serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
+    assert_eq!(parsed.delivery.expect("round trip").skipped, 3);
+
+    // An older file, written before this section existed, must still parse.
+    let legacy: HealthFile = serde_json::from_str(r#"{"ts":1,"sources":{}}"#).unwrap();
+    assert!(legacy.delivery.is_none());
+
+    std::fs::remove_dir_all(&dir).ok();
+}