fix(observability): bound repeated durable-append failure reports

src/graph/observability/README.md

@@ -56,8 +56,11 @@ so existing runs are unchanged.
 async journal API through a background `AppendWorker`: `emit` hands the
 observation to a bounded channel drained on a dedicated thread, so it never
 blocks the executor on I/O. Persistence is **best-effort**: a full queue drops
-(and counts) the overflow, backend errors are reported to stderr rather than
-propagated, and neither aborts the run. The executor calls `GraphEventSink::flush`
+(and counts) the overflow, backend errors are counted and reported through
+`tracing` (rate-limited to one report per failure run plus a reminder every five
+minutes — see the harness `observability` README for the full error policy)
+rather than propagated, and neither aborts the run. The executor calls
+`GraphEventSink::flush`
 after the terminal run event (and callers can call it directly) to block until
 the durable log has caught up. Do not rely on the sink for delivery guarantees
 stronger than "usually persisted, never run-blocking."

src/harness/observability/README.md

@@ -56,11 +56,25 @@ Persisting sinks (`JournalSink`, `JsonlSink`) bridge the synchronous
 background `AppendWorker`: `on_event` hands the observation to a **bounded**
 channel drained on a dedicated thread, so it never blocks the run on I/O.
 Persistence is **best-effort**: if the queue is full the observation is dropped
-(and counted), backend errors are reported to stderr rather than propagated,
-and neither ever aborts the run. Call `JournalSink::flush` / `JsonlSink::flush`
-to block until the durable log has caught up (for example before reading it back
-or shutting down). Do not rely on a sink for delivery guarantees stronger than
-"usually persisted, never run-blocking."
+(and counted in `AppendWorker::dropped`), backend errors are counted (in
+`AppendWorker::append_failures`) and reported through `tracing` rather than
+propagated, and neither ever aborts the run. Call `JournalSink::flush` /
+`JsonlSink::flush` to block until the durable log has caught up (for example
+before reading it back or shutting down). Do not rely on a sink for delivery
+guarantees stronger than "usually persisted, never run-blocking."
+
+A persistent backend failure (read-only volume, full disk) fails every queued
+observation identically, so reporting is **rate-limited** rather than
+per-observation: the first failure of a run logs at `ERROR` on the
+`tinyagents::observability` target, further failures are counted silently with a
+`WARN` reminder at most once per `APPEND_REPORT_COOLDOWN` (5 minutes), and the
+first success afterwards logs one `WARN` recovery summary carrying how many
+observations were lost. The worker keeps attempting every item while degraded —
+the attempt is what detects recovery. Because reporting goes through `tracing`,
+an embedder with no subscriber installed sees nothing — install a subscriber to
+observe durable-log loss. `AppendWorker::append_failures` counts it, but like
+the queue-full `dropped` count it is crate-internal, so it is not a signal a
+host application can read today.
 
 ## Latency metrics semantics
 

src/harness/observability/test.rs

@@ -4,6 +4,7 @@
 
 use std::sync::Arc;
 
+use crate::error::TinyAgentsError;
 use crate::harness::events::{AgentEvent, EventListener, EventRecord, HarnessRunStatus, LimitKind};
 use crate::harness::ids::{CallId, ComponentId, EventId, ExecutionStatus, RunId, ThreadId};
 use crate::harness::observability::AppendWorker;
@@ -362,6 +363,112 @@ async fn append_worker_drops_and_counts_when_queue_is_full() {
     );
 }
 
+#[tokio::test]
+async fn append_worker_counts_failed_appends() {
+    // A sink that rejects every append: each failure is counted, and that count
+    // is kept separate from the queue-full drop count (nothing was dropped —
+    // every item reached the sink and was rejected by it).
+    let worker = AppendWorker::spawn("test-failing", 64, move |_n: u64| async move {
+        Err(TinyAgentsError::Storage("sink offline".into()))
+    });
+
+    for n in 0..12 {
+        worker.submit(n);
+    }
+    worker.flush();
+
+    // Both loss counters are surfaced on the debug view, so an operator dumping
+    // a sink can tell "never reached the backend" from "backend rejected it".
+    let debug = format!("{worker:?}");
+    assert!(
+        debug.contains("append_failures: 12") && debug.contains("dropped: 0"),
+        "debug view must distinguish append failures from queue-full drops, got {debug}"
+    );
+
+    assert_eq!(
+        worker.append_failures(),
+        12,
+        "every failed durable append must be counted"
+    );
+    assert_eq!(
+        worker.dropped(),
+        0,
+        "append failures must not be conflated with queue-full drops"
+    );
+}
+
+#[tokio::test]
+async fn append_worker_failure_then_recovery_keeps_attempting() {
+    use std::sync::Mutex;
+
+    // The sink fails the first 3 appends, then starts accepting. A worker that
+    // stopped attempting while degraded would never persist anything after the
+    // failure run; keeping attempts is what detects the recovery.
+    let attempts = Arc::new(std::sync::atomic::AtomicU64::new(0));
+    let seen: Arc<Mutex<Vec<u64>>> = Arc::new(Mutex::new(Vec::new()));
+    let counter = Arc::clone(&attempts);
+    let sink = Arc::clone(&seen);
+    // A zero cooldown makes the "still failing" reminder fire on every failure
+    // after the first, exercising the suppression path without a wall clock.
+    let worker = AppendWorker::spawn_with_cooldown(
+        "test-flaky",
+        64,
+        std::time::Duration::ZERO,
+        move |n: u64| {
+            let counter = Arc::clone(&counter);
+            let sink = Arc::clone(&sink);
+            async move {
+                if counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed) < 3 {
+                    return Err(TinyAgentsError::Storage("sink offline".into()));
+                }
+                sink.lock().unwrap().push(n);
+                Ok(())
+            }
+        },
+    );
+
+    for n in 0..8 {
+        worker.submit(n);
+    }
+    worker.flush();
+
+    assert_eq!(
+        worker.append_failures(),
+        3,
+        "only the appends the sink rejected are counted as failures"
+    );
+    assert_eq!(
+        *seen.lock().unwrap(),
+        (3..8).collect::<Vec<_>>(),
+        "the worker must keep attempting while degraded, so items persist once the sink recovers"
+    );
+}
+
+#[test]
+fn should_report_bounds_repeats_to_one_per_cooldown() {
+    use crate::harness::observability::worker::should_report;
+    use std::time::{Duration, Instant};
+
+    let cooldown = Duration::from_secs(300);
+    let start = Instant::now();
+
+    // The first failure of a run has never reported, so it always reports.
+    assert!(should_report(None, start, cooldown));
+    // A failure inside the cooldown window is suppressed (counted, not logged).
+    assert!(!should_report(
+        Some(start),
+        start + Duration::from_secs(299),
+        cooldown
+    ));
+    // Once the cooldown has elapsed, the run is due for a reminder again.
+    assert!(should_report(Some(start), start + cooldown, cooldown));
+    assert!(should_report(
+        Some(start),
+        start + Duration::from_secs(600),
+        cooldown
+    ));
+}
+
 /// Collects forwarded records for assertions.
 struct Collector {
     records: std::sync::Mutex<Vec<EventRecord>>,

src/harness/observability/worker.rs

@@ -19,9 +19,37 @@
 //! the run). Callers that need a lossless log can inspect the dropped count.
 //!
 //! # Error policy
-//! Append errors are **not** silently discarded: the drain loop reports each
-//! failure to stderr with the sink name. Persistence remains best-effort — an
-//! error never propagates back into the run — but it is observable.
+//! Append errors are **not** silently discarded, but neither are they allowed to
+//! flood the host's log. Every failed append is counted in
+//! [`AppendWorker::append_failures`] (a lifetime total, distinct from the
+//! queue-full [`AppendWorker::dropped`] count), and the drain loop reports
+//! through `tracing` on the `tinyagents::observability` target with a `sink`
+//! field:
+//!
+//! - the **first** failure of a failure run is reported at `ERROR` — durable
+//!   observations are being lost;
+//! - subsequent failures are counted silently, with a reminder at `WARN` at most
+//!   once per [`APPEND_REPORT_COOLDOWN`]; each reminder carries the *latest*
+//!   error, so a changed cause (read-only volume → full disk) still surfaces
+//!   within one cooldown;
+//! - the first success after a failure run emits one `WARN` recovery summary
+//!   carrying how many observations were lost while the sink was down;
+//! - if the worker shuts down while still failing, it emits one final `WARN`
+//!   summary so a never-recovering run is not silently quiet.
+//!
+//! The worker keeps attempting every item while degraded: the attempt *is* the
+//! recovery detector, it runs off the run's critical path, and skipping it would
+//! turn a transient blip into guaranteed loss of everything still queued. Items
+//! that fail are lost, but counted.
+//!
+//! Persistence remains best-effort — an error never propagates back into the run
+//! — but it is observable. Note that reporting goes through `tracing`, so an
+//! embedder with no subscriber installed sees nothing at all: installing one is
+//! how a host observes durable-log loss. [`AppendWorker::append_failures`]
+//! counts it, but — like the queue-full [`AppendWorker::dropped`] count it
+//! mirrors — it is crate-internal and is not reachable from a host application
+//! through [`JournalSink`](super::JournalSink) or
+//! [`JsonlSink`](super::JsonlSink).
 //!
 //! # Ordering & durability boundary
 //! A single drain thread preserves submit order. [`AppendWorker::flush`] blocks
@@ -35,6 +63,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::mpsc::{Sender, SyncSender, TrySendError, sync_channel};
 use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
 
 use crate::error::Result;
 
@@ -44,6 +73,27 @@ use crate::error::Result;
 /// the drop policy engages, without letting the queue grow without bound.
 pub(crate) const DEFAULT_DRAIN_CAPACITY: usize = 1024;
 
+/// Minimum interval between repeated reports of an *ongoing* append-failure run.
+///
+/// A persistent sink failure (a volume flipped read-only, a full disk) fails
+/// every queued observation identically; without a cooldown that is one log line
+/// per arriving event. The first failure is always reported; after that the
+/// drain loop stays quiet for this long between reminders.
+pub(crate) const APPEND_REPORT_COOLDOWN: Duration = Duration::from_secs(300);
+
+/// Whether an ongoing failure run is due for a reminder report.
+///
+/// `last` is when this run last reported (`None` before its first report).
+/// Pure and clock-free so the cooldown policy is unit-testable without a
+/// subscriber or a fake clock.
+pub(super) fn should_report(last: Option<Instant>, now: Instant, cooldown: Duration) -> bool {
+    match last {
+        None => true,
+        // Saturating: a non-monotonic `now` yields zero, not a panic.
+        Some(previous) => now.saturating_duration_since(previous) >= cooldown,
+    }
+}
+
 /// Messages carried over the drain channel.
 enum Msg<T> {
     /// A payload to persist.
@@ -63,6 +113,8 @@ pub(crate) struct AppendWorker<T: Send + 'static> {
     tx: Option<SyncSender<Msg<T>>>,
     /// Count of payloads dropped because the queue was full (or disconnected).
     dropped: Arc<AtomicU64>,
+    /// Lifetime count of payloads whose durable append returned an error.
+    append_failures: Arc<AtomicU64>,
     /// Handle to the drain thread, joined on drop.
     handle: Option<JoinHandle<()>>,
     /// Human-readable sink name used in error reports.
@@ -76,11 +128,31 @@ impl<T: Send + 'static> AppendWorker<T> {
     /// runtime; it returns the async append future. `name` labels the drain
     /// thread and error reports.
     pub(crate) fn spawn<F, Fut>(name: &'static str, capacity: usize, append: F) -> Self
+    where
+        F: Fn(T) -> Fut + Send + 'static,
+        Fut: Future<Output = Result<()>>,
+    {
+        Self::spawn_with_cooldown(name, capacity, APPEND_REPORT_COOLDOWN, append)
+    }
+
+    /// Spawns a drain worker with an explicit failure-report cooldown.
+    ///
+    /// Same as [`Self::spawn`], which supplies [`APPEND_REPORT_COOLDOWN`]. The
+    /// cooldown is a parameter so the suppression and reminder paths can be
+    /// exercised deterministically, without waiting on wall-clock time.
+    pub(crate) fn spawn_with_cooldown<F, Fut>(
+        name: &'static str,
+        capacity: usize,
+        cooldown: Duration,
+        append: F,
+    ) -> Self
     where
         F: Fn(T) -> Fut + Send + 'static,
         Fut: Future<Output = Result<()>>,
     {
         let (tx, rx) = sync_channel::<Msg<T>>(capacity.max(1));
+        let append_failures = Arc::new(AtomicU64::new(0));
+        let failures = Arc::clone(&append_failures);
         let handle = std::thread::Builder::new()
             .name(format!("tinyagents-{name}-drain"))
             .spawn(move || {
@@ -96,24 +168,72 @@ impl<T: Send + 'static> AppendWorker<T> {
                     }
                 };
                 rt.block_on(async move {
+                    // Failure-run state: how many appends have failed since the
+                    // last success, and when this run last reported. Local to
+                    // the drain thread — no shared state, no timer tasks.
+                    let mut failure_run: u64 = 0;
+                    let mut last_report: Option<Instant> = None;
                     while let Ok(msg) = rx.recv() {
                         match msg {
-                            Msg::Item(item) => {
-                                if let Err(e) = append(item).await {
-                                    eprintln!("tinyagents: {name} durable append failed: {e}");
+                            Msg::Item(item) => match append(item).await {
+                                Ok(()) => {
+                                    if failure_run > 0 {
+                                        tracing::warn!(
+                                            target: "tinyagents::observability",
+                                            sink = name,
+                                            lost = failure_run,
+                                            "[observability] durable append recovered; {failure_run} observation(s) lost while the sink was failing"
+                                        );
+                                        failure_run = 0;
+                                        last_report = None;
+                                    }
                                 }
-                            }
+                                Err(error) => {
+                                    failures.fetch_add(1, Ordering::Relaxed);
+                                    failure_run += 1;
+                                    let now = Instant::now();
+                                    if failure_run == 1 {
+                                        last_report = Some(now);
+                                        tracing::error!(
+                                            target: "tinyagents::observability",
+                                            sink = name,
+                                            error = %error,
+                                            "[observability] durable append failed; the observation is lost and repeats are suppressed until the sink recovers"
+                                        );
+                                    } else if should_report(last_report, now, cooldown) {
+                                        last_report = Some(now);
+                                        tracing::warn!(
+                                            target: "tinyagents::observability",
+                                            sink = name,
+                                            error = %error,
+                                            failures = failure_run,
+                                            "[observability] durable append failed; still failing after {failure_run} consecutive observations"
+                                        );
+                                    }
+                                }
+                            },
                             Msg::Flush(ack) => {
                                 let _ = ack.send(());
                             }
                         }
                     }
+                    // The channel closed mid-failure: report once on the way out
+                    // so a run that never recovered is not silently quiet.
+                    if failure_run > 0 {
+                        tracing::warn!(
+                            target: "tinyagents::observability",
+                            sink = name,
+                            lost = failure_run,
+                            "[observability] durable append failed; sink never recovered before shutdown, {failure_run} observation(s) lost"
+                        );
+                    }
                 });
             })
             .expect("spawn durable-drain thread");
         Self {
             tx: Some(tx),
             dropped: Arc::new(AtomicU64::new(0)),
+            append_failures,
             handle: Some(handle),
             name,
         }
@@ -141,6 +261,17 @@ impl<T: Send + 'static> AppendWorker<T> {
         self.dropped.load(Ordering::Relaxed)
     }
 
+    /// Returns the number of payloads whose durable append returned an error.
+    ///
+    /// Distinct from [`Self::dropped`]: those never reached the sink, these were
+    /// attempted and rejected. Reporting is `tracing`-based and suppressed while
+    /// a failure run continues, so this counter is the only subscriber-free
+    /// signal of durable-log loss.
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub(crate) fn append_failures(&self) -> u64 {
+        self.append_failures.load(Ordering::Relaxed)
+    }
+
     /// Blocks until every payload submitted before this call has been persisted.
     pub(crate) fn flush(&self) {
         let Some(tx) = self.tx.as_ref() else {
@@ -160,6 +291,10 @@ impl<T: Send + 'static> fmt::Debug for AppendWorker<T> {
         f.debug_struct("AppendWorker")
             .field("name", &self.name)
             .field("dropped", &self.dropped.load(Ordering::Relaxed))
+            .field(
+                "append_failures",
+                &self.append_failures.load(Ordering::Relaxed),
+            )
             .finish_non_exhaustive()
     }
 }