Tri-thrust federation hardening + domain generalization + ingest caps

.claude/rules/path-2-stream-p.md

@@ -11,7 +11,7 @@ paths:
 
 # Path 2 — Stream P Supporting (IONe)
 
-Work in this repo supports **Path 2 — Modular Infrastructure Risk Intelligence**, the selected growth path for Morton Analytics LLC (decided 2026-05-12). IONe's role: **integration fabric for Morton Analytics' polyglot client-app portfolio.** GroundPulse is the first instance; TerraYield (`../eo_ag/`) and bearingLineDash (`../bearingLineDash/`) are the next reference apps the substrate must accommodate. Full plan: `../morton-analytics-web/md/strategy/path-2-90day-plan.md`. Substrate spec: `md/design/ione-substrate.md`.
+Work in this repo supports **Path 2 — Modular Infrastructure Risk Intelligence**, the selected growth path for Morton Analytics LLC (decided 2026-05-12). IONe's role: **integration fabric for Morton Analytics' polyglot client-app portfolio.** GroundPulse is the first instance; TerraYield (`../eo_ag/`) and bearingLineDash (`../bearingLineDash/`) are the next reference apps the substrate must accommodate. Full plan: `../morton-command-center/strategy/path-2-90day-plan.md`. Substrate spec: `md/design/ione-substrate.md`.
 
 ## Positioning inside Path 2
 
@@ -40,7 +40,7 @@ P7 is Tier 3 — protected against P2 (pipelines module) but cuttable if Stream
 ## Working in this repo
 
 - All Stream P commits should reference the outcome ID (currently P7) and the GroundPulse-substrate framing.
-- Major architectural decisions go in `md/design/`. Cross-reference the Path 2 tracker (`../morton-analytics-web/md/strategy/path-2-tracker.md`).
+- Major architectural decisions go in `md/design/`. Cross-reference the Path 2 tracker (`../morton-command-center/strategy/path-2-tracker.md`).
 - IONe pricing strategy at `md/strategy/market/ione-pricing.md` is canonical for any pricing discussion — do not re-derive.
 
 ## What this rule prohibits
@@ -58,4 +58,4 @@ P7 is Tier 3 — protected against P2 (pipelines module) but cuttable if Stream
 
 ## When in doubt
 
-Read the 90-day plan at `../morton-analytics-web/md/strategy/path-2-90day-plan.md`. The current week's Tier 1 Stream P outcomes (P1–P3) are the priority. IONe (P7) is supporting.
+Read the 90-day plan at `../morton-command-center/strategy/path-2-90day-plan.md`. Stream P Tier 1 outcomes (P1–P3) shipped; per the replan, IONe (P7) is deferred Tier 3 (target Aug 2026) behind GTM and engine reliability.

CLAUDE.md

@@ -0,0 +1,42 @@
+# CLAUDE.md — IONe
+
+## What this is
+IONe is a **pre-alpha** single Rust binary (axum + tokio + sqlx/Postgres 16 +
+pgvector) with a static UI (`static/`), serving as the MCP-based federation
+substrate underneath GroundPulse and TerraYield: brokered identity (OIDC/SAML),
+signed webhook ingress, generator↔adversarial LLM loop with human approval +
+audit, and a ref-rendering UX shell. It is **OSS engineering infrastructure,
+not a standalone product** — no standalone GTM.
+
+Canonical design: `md/design/ione-substrate.md` (integration-fabric framing).
+Strategy context: `.claude/rules/path-2-stream-p.md`.
+App-integration contract: `md/design/app-integration-playbook.md`.
+
+## Quickstart
+```bash
+cp .env.example .env          # sets IONE_SEED_DEMO=1 (demo workspace)
+docker compose up -d postgres minio
+cargo sqlx database create
+cargo sqlx migrate run
+cargo run --release           # http://localhost:3000
+```
+Two-node federation demo: `./scripts/demo.sh` (nodes on :3000 and :3001).
+
+## Tests
+```bash
+# Cheap unit path:
+cargo test --test phase01_chat
+
+# Integration: live Postgres, serial, #[ignore]-gated:
+DATABASE_URL=postgres://ione:ione@localhost:5433/ione \
+  IONE_SKIP_LIVE=1 \
+  cargo test -- --ignored --test-threads=1
+
+# Playwright e2e:
+npm run test:e2e
+```
+Unset `IONE_SKIP_LIVE` to exercise live Ollama generator/critic/router paths.
+
+## Secrets
+`IONE_TOKEN_KEY` / `IONE_WEBHOOK_SECRET_KEY` live in `.env` (gitignored).
+Never commit `.env` or embed key values in settings/permissions entries.

md/intake/tri-thrust-advance-2026-06-21.md

@@ -0,0 +1,65 @@
+# Intake — Tri-Thrust Advance (2026-06-21)
+
+Canonical work-item ledger from the `ione-tri-thrust` dynamic workflow
+(4 rounds, 45 survivors → deduped below) plus direct code verification.
+Every item has an explicit disposition: **Shipped / Scheduled / Deferred / Out-of-scope**.
+
+Thrusts: **A** = federation data infrastructure · **B** = UI/UX generalization · **C** = AI-augmented ingestion.
+
+## Shipped this session (landed on working tree, `cargo check`+`clippy`+`fmt` green)
+
+| ID | Item | Disposition | Evidence |
+|----|------|-------------|----------|
+| TT-A01 | SQL injection in critic evidence query → parameterized `= ANY($1)` over parsed `Vec<Uuid>` | **Shipped** | `src/services/critic.rs:175` |
+| TT-A02 | Reject bindings/subscriptions to non-active peers (route guard + repo `p.status='active'` backstop) | **Shipped** | `src/routes/peers.rs:291`, `src/repos/workspace_peer_binding_repo.rs:233` |
+| TT-A03 | Re-validate peer + binding status at approval **execution** time | **Shipped** (core guard) | `src/services/federation.rs:357` |
+| TT-A04 | `peers:manage` RBAC gate on create/patch/delete/refresh binding routes | **Shipped** (backend) | `src/routes/bindings.rs` ×4 |
+| TT-A05 | TOCTOU race in MCP session init → atomic DashMap `entry()` | **Shipped** | `src/connectors/peer_session.rs:57` |
+| TT-C09 | 100 KB domain-agnostic cap on webhook `data` (flows into critic LLM prompt) | **Shipped** | `src/routes/webhooks.rs:236` |
+| TT-A06 | `mcp_client` poll fails closed when no Active binding (was unscoped peer-wide enumeration → C-1 cross-workspace read) | **Shipped** | `src/connectors/mcp_client.rs:316` |
+| TT-A07 | Defer `subscribe_peer` first-poll until binding is `Active` (`firstPollDeferred`) | **Shipped** | `src/routes/peers.rs:322` |
+| TT-C01 | `stream_event` payload size cap (64 KiB) at repo choke point; `InsertOutcome::Rejected` + `skipped` in poll response; env `IONE_MAX_STREAM_EVENT_BYTES` | **Shipped** | `src/repos/stream_event_repo.rs:10` |
+| TT-C08 | `bucket_expr` → `Option` (no panic on unvalidated bucket) in stream + audit aggregate repos; `minute` arm added | **Shipped** | `src/repos/{stream_event,audit_event}_aggregate_repo.rs` |
+| TT-B01 | Event-detail panel renders operator-declared `propertyFields` (kills earthquake/PAGER hardcode); `EventLayer.propertyFields` manifest | **Shipped** | `src/services/event_layers.rs:48`, `static/app.js` openEventPopup; req `event-view-schema.md` |
+
+Commits: `b4e2fef` (A-cluster), `003a2fd` (fail-closed/ingest/bucket), `b4ecb88` (B01) on `feat/tri-thrust-federation-hardening`.
+Verification ceiling this session: `cargo check` + `clippy` + `fmt` + `node --check` green. **Integration tests not run** (no Postgres reachable).
+
+## Deferred (real but need design / intent / migration before code)
+
+| ID | Item | Re-entry gate | Source |
+|----|------|---------------|--------|
+| TT-A03b | DENY `interaction_event` trail + UI toast for execution-time revalidation (TT-A03 guard already shipped) | Design: `execute_pending_tool_call` has no `AuthContext`, so the InteractionEvent caller-attribution contract needs a decision before emitting. Error already surfaces to the user today. | #6, #15 |
+| TT-A04b | UI probe-and-hide for binding mutation buttons (pairs TT-A04 backend gate) | Design: no `peers:manage`-gated GET exists to probe, and `/me` per-workspace permission scoping isn't wired for it. Backend gate is the control; 403-on-click is already handled by the error toast. | #1 |
+| TT-C02 | Unescaped user data (signal titles, stream names) inlined into LLM prompts | Design: prompt-injection mitigation approach (delimiting vs. structured prompts vs. sanitization) is not a surgical edit; needs a decision. | #8 |
+| TT-A08 | Pending-tool-call dedup excludes `executed` rows | **Owner intent**: is repeat-execution with identical args intended? If dedup desired → new migration + DB verification | #16, #29 |
+| TT-A09 | Advisory lock in approval execution doesn't cover subsequent DB ops | Design: widen to a transaction or extend lock scope | #17 |
+| TT-A10 | Peer-session task survives peer deletion (orphan) | Design: tie session lifecycle to peer delete | #28 |
+| TT-B02 | Connector provider tiles/config backend-driven (kills Fire/Weather/Pipeline hardcode) | Design: backend config + discovery endpoint | #18,#23,#30,#36,#40 |
+| TT-B03 | Panel-visibility contract decoupled from geospatial fields (data-presence nav) | Design: generalize `GET /workspaces/:id` presence counts | #32, #43 |
+| TT-B04 | Generalize `eventLayers` naming geospatial→generic | Design: API rename, coordinate w/ TT-B01 | #37, #38 |
+| TT-C03 | LLM stream-schema inference / validation-rule generation [L] | Design: validate against Postgres + MCP surface (no Iceberg/Git) | #10 |
+| TT-C04 | LLM evalexpr rule authoring from rule diagnostics | Design | #24 |
+| TT-C05 | Interaction-event intent/context classification at batch ingest | Design | #33 |
+| TT-C06 | Generator stream selection deterministic first-only | Design: selection strategy | #39 |
+| TT-C07 | stream-event field-presence/cardinality inference (feeds TT-C03) | Design | #44 |
+
+## Out-of-scope
+
+| ID | Item | Justification |
+|----|------|---------------|
+| TT-B05 | Demo chat prompts hardcoded fire/weather | Demo fixture, not product surface; lowest leverage (1.67–2.0); regenerate when demo content is revisited | 
+| TT-X01 | Webhook `data` **schema-conformance** validation | A domain-agnostic substrate cannot assume a domain schema; the defensible part (size cap) shipped as TT-C09. Per-peer schema is a peer-app concern, not IONe's. |
+| TT-X02 | Stream-events **external** ingestion size guard | No external POST path; stream events are internal-only. Residual malicious-peer-poll risk covered by TT-C01 (repo-level cap). |
+
+## Coverage audit
+
+All 45 workflow survivors map to an ID above (deduped). No silent drops:
+- Thrust A (20 survivors) → TT-A01..A10 + TT-C09 (webhook). Shipped 5, Scheduled 4, Deferred 3.
+- Thrust B (15 survivors) → TT-B01..B05. Scheduled 1, Deferred 3, Out-of-scope 1.
+- Thrust C (10 survivors) → TT-C01..C08. Shipped 1, Scheduled 3, Deferred 5, Out-of-scope 1.
+- `Unvalidated evidence query pattern in critic` (#9,#25) resolved by TT-A01; residual `bucket_expr` panic → TT-C08.
+
+Every backend item has a paired UI task or an explicit "backend-only" / "Out-of-scope" note.
+</content>
+</invoke>

md/requirements/active/event-view-schema.md

@@ -0,0 +1,40 @@
+# Requirements — Event View Schema (Domain-Agnostic Detail Rendering)
+
+**Source:** `md/intake/tri-thrust-advance-2026-06-21.md` (TT-B01)
+**Status:** code-complete on `feat/tri-thrust-federation-hardening` (Partial — pending founder walkthrough)
+
+The thin UX shell must render event detail without assuming any domain schema. The map
+detail panel previously hardcoded USGS earthquake field names (magnitude, depth, PAGER
+alert, place, USGS url). This couples the substrate to one domain and violates the
+"pluggable view types" layer (`.claude/rules/path-2-stream-p.md`).
+
+## Contract
+
+`GET /api/v1/workspaces/:id/event-layers` — each `EventLayer` carries:
+
+| Field | Type | Meaning |
+|---|---|---|
+| `propertyFields` | `string[]` | Ordered, operator-declared property field names present on every feature's `properties`. Derived from `view_config.property_fields[].name`. |
+
+Each feature's `properties` already contains exactly those named fields (resolved from the
+configured JSON pointers) plus the internal `_event_id` / `_observed_at` keys.
+
+## Rendering rules (frontend)
+
+- The detail panel (`openEventPopup`, `static/app.js`) renders **only** the fields named in
+  `layer.propertyFields`, in declared order, reading values from `feature.properties`.
+- A value matching `^https?://` renders as a safe external link (`rel="noopener noreferrer"`);
+  all other values render as escaped text. No domain-specific field handling (no magnitude,
+  depth, or PAGER chip logic).
+- Title comes from `eventFeatureLabel` (operator `style.labelField`, else first non-internal
+  property, else `_event_id` / stream name).
+- Fallback: when a layer carries no `propertyFields` manifest, render all non-internal
+  (`_`-prefixed excluded) properties so older payloads still display.
+
+## Out of scope
+
+- Per-field display labels and format hints (`PropertyFormat`) — deferred; operators set a
+  friendly `name` today. Revisit if a field needs a label distinct from its key.
+- Dead `pager-chip` CSS in `static/style.css` is now unused; left in place (harmless) for a
+  separate cleanup pass.
+</content>

src/connectors/mcp_client.rs

@@ -232,7 +232,7 @@ impl ConnectorImpl for McpClientConnector {
 
         // list_survivors and search_stream_events require workspace_id.
         // Resolve all workspace ids and aggregate results.
-        let workspace_ids = self.resolve_workspace_ids_with_binding().await;
+        let workspace_ids = self.resolve_workspace_ids_with_binding().await?;
         let now = chrono::Utc::now();
         let mut all_events = Vec::new();
 
@@ -313,27 +313,37 @@ impl McpClientConnector {
         }
     }
 
-    async fn resolve_workspace_ids_with_binding(&self) -> Vec<String> {
+    async fn resolve_workspace_ids_with_binding(&self) -> anyhow::Result<Vec<String>> {
+        // When this connector is bound to a workspace+peer, the poll scope MUST come
+        // from an Active binding's foreign_workspace_id. Falling back to the peer-wide
+        // unscoped enumeration here would read across workspaces the binding does not
+        // authorize (C-1 workspace-isolation leak), so this path fails closed.
         if let (Some(pool), Some(workspace_id), Some(peer_id)) =
             (&self.pool, self.workspace_id, self.peer_id)
         {
-            match WorkspacePeerBindingRepo::new(pool.clone())
+            return match WorkspacePeerBindingRepo::new(pool.clone())
                 .get_by_workspace_peer(workspace_id, peer_id)
                 .await
             {
                 Ok(Some(binding)) if binding.status == BindingStatus::Active => {
-                    if let Some(foreign_workspace_id) = binding.foreign_workspace_id {
-                        if !foreign_workspace_id.is_empty() {
-                            return vec![foreign_workspace_id];
-                        }
+                    match binding.foreign_workspace_id {
+                        Some(fw) if !fw.is_empty() => Ok(vec![fw]),
+                        _ => anyhow::bail!(
+                            "mcp_client: active binding for peer {} has no foreign_workspace_id; poll blocked",
+                            peer_id
+                        ),
                     }
                 }
-                Ok(_) => {}
-                Err(e) => warn!("mcp_client: binding lookup failed during poll: {}", e),
-            }
+                Ok(_) => anyhow::bail!(
+                    "mcp_client: no active workspace_peer_binding for peer {}; poll blocked (fail-closed)",
+                    peer_id
+                ),
+                Err(e) => Err(e.context("mcp_client: binding lookup failed during poll")),
+            };
         }
 
-        self.resolve_all_peer_workspace_ids().await
+        // Unbound connector (no workspace/peer context): legacy peer-wide resolution.
+        Ok(self.resolve_all_peer_workspace_ids().await)
     }
 
     /// Resolve the peer's first workspace id via tools/call list_workspaces.

src/connectors/peer_session.rs

@@ -55,9 +55,11 @@ impl PeerSessionRegistry {
     }
 
     pub fn start(&self, state: AppState, peer_id: Uuid) {
-        if self.tasks.contains_key(&peer_id) {
-            return;
-        }
+        use dashmap::mapref::entry::Entry;
+        let vacant = match self.tasks.entry(peer_id) {
+            Entry::Occupied(_) => return,
+            Entry::Vacant(vacant) => vacant,
+        };
         let (state_tx, _state_rx) = watch::channel(SessionState::Connecting);
         let tx = state_tx.clone();
         let limit = self.limit.clone();
@@ -68,8 +70,7 @@ impl PeerSessionRegistry {
             };
             run_session_task(state, peer_id, tx).await;
         });
-        self.tasks
-            .insert(peer_id, PeerSessionHandle { state_tx, task });
+        vacant.insert(PeerSessionHandle { state_tx, task });
     }
 
     pub fn stop(&self, peer_id: Uuid) {

src/repos/audit_event_aggregate_repo.rs

@@ -47,10 +47,14 @@ pub struct ActorCountRow {
 
 /// `bucket` must be pre-validated by the route against {minute,hour,day,week};
 /// it is interpolated into date_trunc. Fail loudly rather than silently default.
-fn bucket_expr(bucket: &str) -> String {
+fn bucket_expr(bucket: &str) -> Option<String> {
+    // Last allow-list gate before `bucket` is interpolated into SQL; return None
+    // (→ caller error) rather than panicking on a value that slips past the route layer.
     match bucket {
-        "minute" | "hour" | "day" | "week" => format!("date_trunc('{bucket}', ae.created_at)"),
-        other => panic!("bucket '{other}' must be validated before bucket_expr"),
+        "minute" | "hour" | "day" | "week" => {
+            Some(format!("date_trunc('{bucket}', ae.created_at)"))
+        }
+        _ => None,
     }
 }
 
@@ -67,7 +71,8 @@ impl AuditEventAggregateRepo {
         group_col: GroupCol,
         filter: &AuditEventFilter,
     ) -> anyhow::Result<Vec<BucketCountRow>> {
-        let bucket_expr = bucket_expr(bucket);
+        let bucket_expr =
+            bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?;
         let key_expr = group_col.key_expr();
         let mut qb = QueryBuilder::new(format!(
             "SELECT {bucket_expr} AS bucket_start, {key_expr} AS key, COUNT(*)::bigint AS count"

src/repos/stream_event_aggregate_repo.rs

@@ -47,7 +47,8 @@ impl StreamEventAggregateRepo {
         until: DateTime<Utc>,
         bucket: &str,
     ) -> anyhow::Result<Vec<Value>> {
-        let bucket_expr = bucket_expr(bucket);
+        let bucket_expr =
+            bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?;
         let sql = format!(
             "SELECT {bucket_expr} AS bucket_start,
                     (EXTRACT(EPOCH FROM {bucket_expr}) * 1000)::bigint AS bucket_start_ms,
@@ -100,7 +101,8 @@ impl StreamEventAggregateRepo {
         bucket: &str,
         value_path: Vec<String>,
     ) -> anyhow::Result<Vec<Value>> {
-        let bucket_expr = bucket_expr(bucket);
+        let bucket_expr =
+            bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?;
         let sql = format!(
             "SELECT {bucket_expr} AS bucket_start,
                     (EXTRACT(EPOCH FROM {bucket_expr}) * 1000)::bigint AS bucket_start_ms,
@@ -163,7 +165,8 @@ impl StreamEventAggregateRepo {
         value_path: Vec<String>,
         pct: f64,
     ) -> anyhow::Result<Vec<Value>> {
-        let bucket_expr = bucket_expr(bucket);
+        let bucket_expr =
+            bucket_expr(bucket).ok_or_else(|| anyhow::anyhow!("invalid bucket: {bucket}"))?;
         let sql = format!(
             "SELECT {bucket_expr} AS bucket_start,
                     (EXTRACT(EPOCH FROM {bucket_expr}) * 1000)::bigint AS bucket_start_ms,
@@ -317,13 +320,15 @@ impl StreamEventAggregateRepo {
     }
 }
 
-fn bucket_expr(bucket: &str) -> &'static str {
+fn bucket_expr(bucket: &str) -> Option<&'static str> {
+    // `bucket` is interpolated into SQL; this is the last allow-list gate before the
+    // value reaches the query. Return None (→ caller error) rather than panicking, so a
+    // bucket that slips past the route layer can never reach SQL or crash the worker.
     match bucket {
-        "hour" => "date_trunc('hour', se.observed_at)",
-        "day" => "date_trunc('day', se.observed_at)",
-        "week" => "date_trunc('week', se.observed_at)",
-        // `bucket` is interpolated into SQL; callers must validate it against the
-        // allow-list (route layer) first. Fail loudly rather than silently default.
-        other => panic!("bucket '{other}' must be validated before bucket_expr"),
+        "minute" => Some("date_trunc('minute', se.observed_at)"),
+        "hour" => Some("date_trunc('hour', se.observed_at)"),
+        "day" => Some("date_trunc('day', se.observed_at)"),
+        "week" => Some("date_trunc('week', se.observed_at)"),
+        _ => None,
     }
 }