reliability audit remediation: erasure closure, durable state machines, provider budgets, and config hardening

.agents/skills/moa-100-session-sweep/SKILL.md

@@ -99,6 +99,29 @@ Useful env overrides:
 - `MOA_SWEEP_IDS`: comma-separated session ids for focused lanes
 - `MOA_SWEEP_MODEL`: pinned model, usually `gpt-5.4-mini`
 - `MOA_SWEEP_REDIS_URL`: runtime-cache Redis URL, default `redis://127.0.0.1:10051/0`
+- `MOA_SWEEP_PROVIDER_MAX_IN_FLIGHT`: per-credential provider in-flight budget for the
+  sweep orchestrator, default `64`. Chat calls are bounded per provider credential by
+  default (16 unless configured), which a full sweep can saturate; the runner passes this
+  through as `MOA_OPENAI_MAX_CONCURRENT_REQUESTS` so the budget never shapes outcomes.
+  The runner also forces `MOA_PROVIDERS_CONCURRENCY_SCOPE=local` so the sweep never
+  shares Redis lease budgets with the long-running compose orchestrator.
+
+## Runtime-Behavior Notes (defaults changed 2026-07-10)
+
+Interpretation guidance for sweeps at or after the 2026-07-10 defaults changes:
+
+- **Non-builtin (MCP) tools default to admin review.** Sweep personas use builtin tools
+  only, so current cases are unaffected — but any future persona that registers an MCP
+  tool will stall on an approval unless the case seeds an operator allow rule or handles
+  the review flow. Do not read such a stall as a delegation/fan-in regression.
+- **Skill distillation now requires 8+ tool calls per segment** (was 5). This gates
+  skill *learning*, not skill *activation* — `F-SKILL-INJECT` reads persisted
+  `skills_activated` segment evidence and is unaffected. Expect fewer
+  distillation proposals from short sweep sessions; that is intended, not a regression.
+- **Provider chat concurrency is bounded per credential by default.** The runner sizes
+  the sweep budget via `MOA_SWEEP_PROVIDER_MAX_IN_FLIGHT` (see above). If a sweep shows
+  clustered latency spikes or rate-limit failovers, check that override before suspecting
+  a provider or orchestrator regression.
 
 ## Baseline Rules
 

.agents/skills/moa-100-session-sweep/scripts/run_100_session_sweep.py

@@ -39,6 +39,11 @@
 # Minimum contiguous match length (chars) that counts as the final reply leaking raw worker output.
 RAW_LEAK_MIN_CHARS = 120
 MAX_WORKERS = int(os.environ.get("MOA_SWEEP_CONCURRENCY", "4"))
+# Provider in-flight budget for the sweep's single live key. Chat calls are
+# bounded per provider credential by default (16 unless configured), which a
+# full sweep (MOA_SWEEP_CONCURRENCY sessions x coordinator + spawned workers)
+# can saturate; size generously so the budget never shapes sweep outcomes.
+PROVIDER_MAX_IN_FLIGHT = os.environ.get("MOA_SWEEP_PROVIDER_MAX_IN_FLIGHT", "64")
 SESSION_TIMEOUT_S = int(os.environ.get("MOA_SWEEP_SESSION_TIMEOUT_S", "260"))
 TURN_LIMIT = int(os.environ.get("MOA_SWEEP_MAX_TURNS", "6"))
 CASE_LIMIT = int(os.environ.get("MOA_SWEEP_LIMIT", "0"))
@@ -279,6 +284,12 @@ def start_orchestrator(env):
             "MOA_LOCAL_DOCKER_ENABLED": "false",
             "MOA_RUNTIME_CACHE_BACKEND": "redis",
             "MOA_RUNTIME_CACHE_REDIS_URL": SWEEP_REDIS_URL,
+            # The sweep shares the compose Redis; force local concurrency scope
+            # so an inherited global scope can never make the sweep orchestrator
+            # share provider lease budgets with the long-running compose stack.
+            "MOA_PROVIDERS_CONCURRENCY_SCOPE": "local",
+            # Sweep model is routed through the OpenAI provider credential.
+            "MOA_OPENAI_MAX_CONCURRENT_REQUESTS": PROVIDER_MAX_IN_FLIGHT,
             "RUST_LOG": os.environ.get("RUST_LOG", "info,moa_orchestrator=info"),
         }
     )

SEQUENCE-DIAGRAMS.md

@@ -57,7 +57,7 @@ sequenceDiagram
     Pipe-->>Brain: WorkingContext (cache_breakpoint marked)
 
     Brain->>LLM: complete(compiled_context)
-    LLM-->>Brain: ToolCall { bash: "fly deploy --app staging" }
+    LLM-->>Brain: ToolCall { bash: "deploy --app staging" }
 
     Brain->>Log: emit ApprovalRequested
     Orch->>Messaging: observe event
@@ -70,7 +70,7 @@ sequenceDiagram
     Orch->>Brain: deliver signal
     Brain->>Log: emit ApprovalDecided
 
-    Brain->>Router: execute("bash", "fly deploy ...")
+    Brain->>Router: execute("bash", "deploy ...")
     Router->>Hand: provision (lazy, first call)
     Hand-->>Router: HandHandle
     Router->>Hand: execute(bash, input)
@@ -384,7 +384,7 @@ sequenceDiagram
 
     OldBrain->>Log: emit ToolCall
     OldBrain->>Log: emit ToolResult
-    Note over OldBrain: 💥 process crash / machine restart<br/>(panic, OOM, Fly.io suspend, Ctrl+C)
+    Note over OldBrain: 💥 process crash / machine restart<br/>(panic, OOM, host suspend, Ctrl+C)
 
     Note over Orch: On next signal OR startup scan
     Orch->>Log: list_sessions(status in [Running, WaitingApproval])

crates/moa-analytics/Cargo.toml

@@ -19,4 +19,5 @@ workspace-hack.workspace = true
 [dev-dependencies]
 clickhouse = { workspace = true, features = ["test-util"] }
 insta.workspace = true
+moa-test-support.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

crates/moa-analytics/src/clickhouse_exec.rs

@@ -34,8 +34,17 @@ fn is_unknown_table(error: &clickhouse::error::Error) -> bool {
 #[derive(Clone)]
 pub struct AnalyticsClickHouseClient {
     client: Client,
+    max_execution_time_secs: u64,
+    max_rows_to_read: u64,
+    max_bytes_to_read: u64,
 }
 
+/// Default ClickHouse per-query budgets applied when the caller does not
+/// override them from config. Mirror [`moa_core::config::AnalyticsConfig`].
+const DEFAULT_MAX_EXECUTION_TIME_SECS: u64 = 10;
+const DEFAULT_MAX_ROWS_TO_READ: u64 = 1_000_000_000;
+const DEFAULT_MAX_BYTES_TO_READ: u64 = 10_000_000_000;
+
 impl std::fmt::Debug for AnalyticsClickHouseClient {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("AnalyticsClickHouseClient")
@@ -56,13 +65,33 @@ impl AnalyticsClickHouseClient {
         if let Some(password) = config.password.as_deref() {
             client = client.with_password(password);
         }
-        Self { client }
+        Self::from_client(client)
     }
 
     /// Wraps an existing client, used by tests to point at a mock server.
     #[must_use]
     pub fn from_client(client: Client) -> Self {
-        Self { client }
+        Self {
+            client,
+            max_execution_time_secs: DEFAULT_MAX_EXECUTION_TIME_SECS,
+            max_rows_to_read: DEFAULT_MAX_ROWS_TO_READ,
+            max_bytes_to_read: DEFAULT_MAX_BYTES_TO_READ,
+        }
+    }
+
+    /// Overrides the per-query ClickHouse budgets (`max_execution_time`,
+    /// `max_rows_to_read`, `max_bytes_to_read`) so a runaway read is bounded.
+    #[must_use]
+    pub fn with_query_budgets(
+        mut self,
+        max_execution_time_secs: u64,
+        max_rows_to_read: u64,
+        max_bytes_to_read: u64,
+    ) -> Self {
+        self.max_execution_time_secs = max_execution_time_secs;
+        self.max_rows_to_read = max_rows_to_read;
+        self.max_bytes_to_read = max_bytes_to_read;
+        self
     }
 
     /// Deletes a tenant's rows from every analytics table during offboarding.
@@ -107,7 +136,17 @@ impl AnalyticsClickHouseClient {
         &self,
         compiled: &CompiledAnalyticsQuery,
     ) -> Result<Vec<Vec<AnalyticsCell>>> {
-        let mut query = self.client.query(&compiled.sql);
+        // Bound the read: a runaway scan is cancelled by ClickHouse instead of
+        // reading a tenant's full history to satisfy an exact ordered percentile.
+        let mut query = self
+            .client
+            .query(&compiled.sql)
+            .with_option(
+                "max_execution_time",
+                self.max_execution_time_secs.to_string(),
+            )
+            .with_option("max_rows_to_read", self.max_rows_to_read.to_string())
+            .with_option("max_bytes_to_read", self.max_bytes_to_read.to_string());
         for bind in &compiled.bind_values {
             query = match bind {
                 AnalyticsBindValue::String(value) => query.bind(value),

crates/moa-analytics/src/compiler.rs

@@ -729,7 +729,15 @@ mod tests {
                 aggregation: AnalyticsAggregation::P95,
                 alias: Some("p95_cost".to_string()),
             }],
-            filters: Vec::new(),
+            // Time-series datasets require a bounded window; a recent lower bound
+            // keeps the shared fixture valid against the wall clock.
+            filters: vec![AnalyticsFilter {
+                field: "finished_at".to_string(),
+                operator: AnalyticsFilterOperator::Gte,
+                value: Some(AnalyticsCell::String(
+                    (chrono::Utc::now() - chrono::Duration::days(1)).to_rfc3339(),
+                )),
+            }],
             order_by: vec![AnalyticsOrderBy {
                 field: "p95_cost".to_string(),
                 direction: AnalyticsSortDirection::Desc,
@@ -748,7 +756,8 @@ mod tests {
         assert!(compiled.sql.contains("d.tenant_id = $1::UUID"));
         assert!(compiled.sql.contains("PERCENTILE_CONT(0.95)"));
         assert!(compiled.sql.contains("ORDER BY c1 DESC"));
-        assert_eq!(compiled.bind_values.len(), 1);
+        // Tenant scope bind plus the required time-window lower bound.
+        assert_eq!(compiled.bind_values.len(), 2);
         assert_eq!(compiled.limit, 25);
     }
 

crates/moa-analytics/src/error.rs

@@ -120,6 +120,20 @@ pub enum AnalyticsError {
         /// Maximum allowed time window in whole days.
         max_days: i64,
     },
+    /// A time-series dataset was queried without a bounded time window.
+    #[error(
+        "analytics dataset `{dataset}` requires a bounded time filter on `{time_field}` \
+         (a two-sided `between` within {max_days} days, or a lower bound `gte`/`gt`/`eq` \
+         no older than {max_days} days)"
+    )]
+    MissingTimeWindow {
+        /// Dataset that requires a bounded time window.
+        dataset: String,
+        /// Designated time field that must be bounded.
+        time_field: String,
+        /// Maximum allowed time window in whole days.
+        max_days: i64,
+    },
     /// A query tried to order by a field or alias that is not selected.
     #[error("order field `{field}` is not selected by the query")]
     UnknownOrderField {

crates/moa-analytics/src/executor.rs

@@ -21,9 +21,23 @@ use crate::error::{AnalyticsError, Result};
 /// [`AnalyticsService::clickhouse`] compiles and executes against the ClickHouse
 /// read models. The edge selects the constructor by `[clickhouse]` presence and
 /// calls the matching `query` / `query_clickhouse` entrypoint.
-#[derive(Debug, Clone, Default)]
+/// Default Postgres `statement_timeout` applied to each analytics query when
+/// the caller does not override it from config.
+pub const DEFAULT_STATEMENT_TIMEOUT_MS: u64 = 10_000;
+
+#[derive(Debug, Clone)]
 pub struct AnalyticsService {
     compiler: AnalyticsCompiler,
+    statement_timeout_ms: u64,
+}
+
+impl Default for AnalyticsService {
+    fn default() -> Self {
+        Self {
+            compiler: AnalyticsCompiler::default(),
+            statement_timeout_ms: DEFAULT_STATEMENT_TIMEOUT_MS,
+        }
+    }
 }
 
 impl AnalyticsService {
@@ -39,9 +53,18 @@ impl AnalyticsService {
                 analytics_catalog(),
                 AnalyticsBackend::ClickHouse,
             ),
+            statement_timeout_ms: DEFAULT_STATEMENT_TIMEOUT_MS,
         }
     }
 
+    /// Overrides the Postgres `statement_timeout` applied to each query, in
+    /// milliseconds, so a runaway scan is cancelled server-side.
+    #[must_use]
+    pub fn with_statement_timeout_ms(mut self, statement_timeout_ms: u64) -> Self {
+        self.statement_timeout_ms = statement_timeout_ms;
+        self
+    }
+
     /// Returns the backend this service compiles and executes against.
     pub fn backend(&self) -> AnalyticsBackend {
         self.compiler.backend()
@@ -67,6 +90,14 @@ impl AnalyticsService {
         let mut conn = ScopedConn::begin_tenant(pool, compiled.effective_tenant_id)
             .await
             .map_err(|error| AnalyticsError::Execution(error.to_string()))?;
+        // Bound the database work: an unbounded ordered-percentile scan is
+        // cancelled server-side rather than holding a connection open. `SET LOCAL`
+        // scopes the timeout to this transaction only.
+        sqlx::query("SELECT set_config('statement_timeout', $1, true)")
+            .bind(self.statement_timeout_ms.to_string())
+            .execute(conn.as_mut())
+            .await
+            .map_err(|error| AnalyticsError::Execution(error.to_string()))?;
         let rows = execute_compiled(&compiled, conn.as_mut()).await?;
         conn.commit()
             .await