feat(subconscious): event-driven trigger pipeline + orchestrator (#3650)

Co-authored-by: M3gA-Mind <megamind@mahadao.com>

tests/subconscious_conversation_e2e.rs

@@ -0,0 +1,474 @@
+//! Multi-party conversation e2e: **human ↔ subconscious orchestrator ↔ sub-agent**.
+//!
+//! This drives the *real* [`TriggerOrchestrator`] (ingest → dedupe/rate →
+//! gate → priority queue → serial run loop) with two injected seams — a
+//! scripted [`Gate`] and a scripted [`SessionExecutor`] — so we can simulate
+//! many back-and-forth comms patterns deterministically, without an LLM:
+//!
+//! - **human → subconscious**: a `ChannelInboundMessage` is normalized,
+//!   gated, promoted, and run.
+//! - **subconscious → sub-agent**: the scripted session "spawns" a sub-agent
+//!   by emitting a follow-on `SubagentCompleted`/`SubagentFailed` event that
+//!   re-enters the real pipeline.
+//! - **sub-agent → subconscious**: that conclusion is normalized, gated, and
+//!   merged by the session.
+//! - **subconscious → human**: the session calls the real `notify_user`,
+//!   which we capture off the event bus.
+//!
+//! Everything between the two seams is the production code path: event
+//! normalization, the dedupe window, per-source rate limiting, the promotion
+//! gate budget, the priority queue, the serial loop, and the anti-self-trigger
+//! guard. The harness re-injects emitted follow-on events through the same
+//! `ingest` entry point the bus subscriber uses, so cascades are exercised
+//! exactly as in production.
+//!
+//! Run with a visible transcript:
+//! `cargo test --test subconscious_conversation_e2e -- --nocapture`
+
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex as StdMutex, OnceLock};
+use std::time::Duration;
+
+use async_trait::async_trait;
+
+use openhuman_core::core::event_bus::{global, init_global, DomainEvent};
+use openhuman_core::openhuman::subconscious_triggers::types::{
+    GateDecision, Trigger, TriggerPriority, TriggerSource,
+};
+use openhuman_core::openhuman::subconscious_triggers::{
+    Gate, OrchestratorConfig, SessionExecutor, TriggerOrchestrator,
+};
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Shared conversation transcript — the record of "what happened".
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[derive(Clone)]
+struct Transcript(Arc<StdMutex<Vec<String>>>);
+
+impl Transcript {
+    fn new() -> Self {
+        Self(Arc::new(StdMutex::new(Vec::new())))
+    }
+    fn push(&self, line: impl Into<String>) {
+        self.0.lock().unwrap().push(line.into());
+    }
+    fn lines(&self) -> Vec<String> {
+        self.0.lock().unwrap().clone()
+    }
+    fn count(&self, needle: &str) -> usize {
+        self.0
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|l| l.contains(needle))
+            .count()
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Scripted gate — promote/drop driven by trigger source + content rules.
+// ─────────────────────────────────────────────────────────────────────────────
+
+struct ScriptedGate {
+    transcript: Transcript,
+}
+
+#[async_trait]
+impl Gate for ScriptedGate {
+    async fn evaluate(&self, trigger: &Trigger, _now: f64) -> GateDecision {
+        let summary = &trigger.payload.gate_summary;
+        // Routine cron noise is dropped.
+        let drop =
+            matches!(trigger.source, TriggerSource::Cron { .. }) || summary.contains("[ignore]");
+        if drop {
+            self.transcript
+                .push(format!("GATE drop     {}", trigger.display_label));
+            return GateDecision::Drop {
+                acknowledge: false,
+                reason: "routine/no-op".into(),
+            };
+        }
+        // Everything else promotes. Urgent if the human flagged it.
+        let priority = if summary.to_lowercase().contains("urgent") {
+            TriggerPriority::Urgent
+        } else {
+            trigger.priority
+        };
+        self.transcript.push(format!(
+            "GATE promote  {} (prio={})",
+            trigger.display_label,
+            priority.as_str()
+        ));
+        GateDecision::Promote {
+            // Keep the original summary so the session can branch on it.
+            synthesized_summary: summary.clone(),
+            priority,
+            reason: "actionable".into(),
+        }
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Scripted session — models the subconscious agent's behaviour, including
+// spawning sub-agents (via emitted follow-on events) and notifying the human.
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// A follow-on event the scripted session wants to feed back into the
+/// orchestrator (simulating the bus fan-in for sub-agent conclusions).
+type Emitter = Arc<StdMutex<VecDeque<DomainEvent>>>;
+
+struct ScriptedSession {
+    transcript: Transcript,
+    workspace: std::path::PathBuf,
+    emit: Emitter,
+    /// Monotonic id for spawned sub-agent tasks.
+    next_task: Arc<std::sync::atomic::AtomicU64>,
+}
+
+impl ScriptedSession {
+    fn emit(&self, event: DomainEvent) {
+        self.emit.lock().unwrap().push_back(event);
+    }
+    fn spawn_subagent(&self, agent_id: &str, ok: bool) -> String {
+        let n = self
+            .next_task
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let task_id = format!("task-{n}");
+        self.transcript
+            .push(format!("SESSION spawn  sub-agent '{agent_id}' ({task_id})"));
+        let event = if ok {
+            DomainEvent::SubagentCompleted {
+                parent_session: "subconscious:orchestrator".into(),
+                task_id: task_id.clone(),
+                agent_id: agent_id.into(),
+                elapsed_ms: 5,
+                output_chars: 120,
+                iterations: 2,
+            }
+        } else {
+            DomainEvent::SubagentFailed {
+                parent_session: "subconscious:orchestrator".into(),
+                task_id: task_id.clone(),
+                agent_id: agent_id.into(),
+                error: "tool timeout".into(),
+            }
+        };
+        self.emit(event);
+        task_id
+    }
+}
+
+#[async_trait]
+impl SessionExecutor for ScriptedSession {
+    async fn execute(&self, summary: &str, _external: bool) -> Result<String, String> {
+        let s = summary.to_lowercase();
+
+        // Branch on what the promoted turn is about — modelling the
+        // orchestrator agent's decisions. ORDER MATTERS: sub-agent conclusion
+        // summaries mention the agent id ("researcher" ⊃ "research"), so the
+        // conclusion branches must be checked before the delegate branch.
+        if s.contains("failed") {
+            // A sub-agent failed → recover, escalate a retry sub-agent.
+            self.transcript
+                .push("SESSION handle sub-agent FAILURE → retry");
+            self.spawn_subagent("researcher", true);
+            Ok("Retrying after failure.".into())
+        } else if s.contains("completed in") || s.contains("chars of output") {
+            // A sub-agent conclusion came back → merge + tell the human.
+            self.transcript
+                .push("SESSION merge  sub-agent conclusion → notifying human");
+            openhuman_core::openhuman::subconscious::notify_user(
+                self.workspace.clone(),
+                "Your research is ready — here's the summary.",
+                Some("research done"),
+            );
+            self.transcript.push("SESSION notify human (done)");
+            Ok("Merged conclusion; user notified.".into())
+        } else if s.contains("research") || s.contains("prep") || s.contains("deck") {
+            // Human asked for deep work → delegate to a sub-agent.
+            self.transcript
+                .push(format!("SESSION run    (delegating) :: {summary}"));
+            self.spawn_subagent("researcher", true);
+            Ok("Delegated to researcher.".into())
+        } else if s.contains("status") || s.contains("what") {
+            // Simple Q → answer the human directly, no sub-agent.
+            self.transcript
+                .push(format!("SESSION reply  directly :: {summary}"));
+            openhuman_core::openhuman::subconscious::notify_user(
+                self.workspace.clone(),
+                "Here's your status.",
+                None,
+            );
+            Ok("Answered directly.".into())
+        } else {
+            self.transcript.push(format!("SESSION note   :: {summary}"));
+            Ok("Noted.".into())
+        }
+    }
+
+    fn thread_id(&self) -> &str {
+        "subconscious:orchestrator"
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Harness.
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Serializes tests that touch the process-global event bus.
+fn bus_lock() -> std::sync::MutexGuard<'static, ()> {
+    static LOCK: OnceLock<StdMutex<()>> = OnceLock::new();
+    LOCK.get_or_init(|| StdMutex::new(()))
+        .lock()
+        .unwrap_or_else(|p| p.into_inner())
+}
+
+struct Harness {
+    orch: Arc<TriggerOrchestrator>,
+    transcript: Transcript,
+    emit: Emitter,
+    notifications: Arc<StdMutex<Vec<String>>>,
+    _loop: tokio::task::JoinHandle<()>,
+    _sub: openhuman_core::core::event_bus::SubscriptionHandle,
+    _tmp: tempfile::TempDir,
+}
+
+impl Harness {
+    fn new(config: OrchestratorConfig) -> Self {
+        init_global(128);
+        let transcript = Transcript::new();
+        let emit: Emitter = Arc::new(StdMutex::new(VecDeque::new()));
+        let tmp = tempfile::tempdir().expect("tempdir");
+
+        let session = Arc::new(ScriptedSession {
+            transcript: transcript.clone(),
+            workspace: tmp.path().to_path_buf(),
+            emit: Arc::clone(&emit),
+            next_task: Arc::new(std::sync::atomic::AtomicU64::new(1)),
+        });
+        let gate = Arc::new(ScriptedGate {
+            transcript: transcript.clone(),
+        });
+        let orch = Arc::new(TriggerOrchestrator::with_components(session, gate, config));
+
+        // Capture proactive (subconscious → human) deliveries off the bus.
+        let notifications = Arc::new(StdMutex::new(Vec::<String>::new()));
+        let sink = Arc::clone(&notifications);
+        let sub = global().expect("bus").on("conv-e2e-notify", move |event| {
+            let sink = Arc::clone(&sink);
+            let event = event.clone();
+            Box::pin(async move {
+                if let DomainEvent::ProactiveMessageRequested {
+                    source, message, ..
+                } = &event
+                {
+                    if source == "subconscious" {
+                        sink.lock().unwrap().push(message.clone());
+                    }
+                }
+            })
+        });
+
+        let loop_handle = Arc::clone(&orch);
+        let task = tokio::spawn(async move { loop_handle.run_loop().await });
+
+        Self {
+            orch,
+            transcript,
+            emit,
+            notifications,
+            _loop: task,
+            _sub: sub,
+            _tmp: tmp,
+        }
+    }
+
+    /// Feed an inbound event into the orchestrator (as the bus subscriber would).
+    fn ingest(&self, event: DomainEvent) {
+        self.orch.ingest(&event);
+    }
+
+    /// Pump the cascade until quiescent: repeatedly drain follow-on events the
+    /// scripted session emitted back into `ingest`, waiting for the queue +
+    /// gate tasks to settle between rounds. Bounded so a runaway loop fails
+    /// fast instead of hanging.
+    async fn settle(&self) {
+        for _ in 0..200 {
+            // Let in-flight gate tasks + the run loop make progress.
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            let pending: Vec<DomainEvent> = {
+                let mut q = self.emit.lock().unwrap();
+                q.drain(..).collect()
+            };
+            for ev in &pending {
+                self.orch.ingest(ev);
+            }
+            // Quiescent when nothing is queued and nothing new was emitted.
+            if pending.is_empty() && self.orch.queue_depth() == 0 {
+                // One more grace round to catch a just-finished session run
+                // that emitted a trailing event.
+                tokio::time::sleep(Duration::from_millis(15)).await;
+                if self.emit.lock().unwrap().is_empty() && self.orch.queue_depth() == 0 {
+                    return;
+                }
+            }
+        }
+        panic!("conversation did not settle — possible runaway cascade");
+    }
+
+    fn notifications(&self) -> Vec<String> {
+        self.notifications.lock().unwrap().clone()
+    }
+
+    fn print(&self, title: &str) {
+        println!("\n=== {title} ===");
+        for line in self.transcript.lines() {
+            println!("  {line}");
+        }
+        for n in self.notifications() {
+            println!("  >> to human: {n}");
+        }
+    }
+}
+
+fn human_msg(channel: &str, sender: &str, message: &str) -> DomainEvent {
+    DomainEvent::ChannelInboundMessage {
+        event_name: "msg".into(),
+        channel: channel.into(),
+        message: message.into(),
+        sender: Some(sender.into()),
+        reply_target: Some("dm".into()),
+        thread_ts: Some(format!("t-{}", message.len())),
+        raw_data: serde_json::Value::Null,
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Scenario 1 — full round trip: human → session → sub-agent → session → human.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn conversation_human_delegates_then_subagent_reports_back() {
+    let _g = bus_lock();
+    let h = Harness::new(OrchestratorConfig::default());
+
+    // Human asks for deep work.
+    h.ingest(human_msg(
+        "slack",
+        "U1",
+        "can you prep the Q3 research deck?",
+    ));
+    h.settle().await;
+    h.print("scenario 1: delegate → sub-agent → report back");
+
+    // The session delegated (spawned a sub-agent) …
+    assert_eq!(
+        h.transcript.count("spawn  sub-agent"),
+        1,
+        "one sub-agent spawned"
+    );
+    // … the conclusion came back through the real pipeline and was merged …
+    assert_eq!(h.transcript.count("merge  sub-agent conclusion"), 1);
+    // … and the human was notified exactly once.
+    let notes = h.notifications();
+    assert_eq!(notes.len(), 1, "exactly one human notification");
+    assert!(notes[0].contains("research is ready"));
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Scenario 2 — sub-agent failure → recovery → retry → success → human.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn conversation_subagent_failure_recovers_with_retry() {
+    let _g = bus_lock();
+    let h = Harness::new(OrchestratorConfig::default());
+
+    // Inject a sub-agent FAILURE conclusion directly (as if a prior spawn failed).
+    h.ingest(DomainEvent::SubagentFailed {
+        parent_session: "subconscious:orchestrator".into(),
+        task_id: "task-0".into(),
+        agent_id: "researcher".into(),
+        error: "tool timeout".into(),
+    });
+    h.settle().await;
+    h.print("scenario 2: failure → retry → success → human");
+
+    // Failure handled, a retry sub-agent spawned, its success merged, human told.
+    assert_eq!(h.transcript.count("handle sub-agent FAILURE"), 1);
+    assert_eq!(h.transcript.count("spawn  sub-agent"), 1, "one retry spawn");
+    assert_eq!(h.transcript.count("merge  sub-agent conclusion"), 1);
+    assert_eq!(h.notifications().len(), 1);
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Scenario 3 — interleaved traffic: cron noise dropped, two humans, dedupe.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn conversation_interleaved_traffic_is_handled() {
+    let _g = bus_lock();
+    let h = Harness::new(OrchestratorConfig::default());
+
+    // Routine cron tick — gate should drop it (no session run).
+    h.ingest(DomainEvent::CronJobTriggered {
+        job_id: "nightly".into(),
+        job_name: "nightly recap".into(),
+        job_type: "agent".into(),
+    });
+    // Human A asks a simple question → direct reply, no sub-agent.
+    h.ingest(human_msg("slack", "U1", "what's my status today?"));
+    // Human B duplicate-sends the exact same message twice (transport retry).
+    let dup = human_msg("discord", "U2", "please prep the deck");
+    h.ingest(dup.clone());
+    h.ingest(dup);
+    h.settle().await;
+    h.print("scenario 3: interleaved cron + two humans + dedupe");
+
+    // Cron was dropped (no session run for it).
+    assert_eq!(h.transcript.count("GATE drop"), 1);
+    // Human A answered directly (no sub-agent for a status question).
+    assert_eq!(h.transcript.count("reply  directly"), 1);
+    // Human B's duplicate collapsed → only ONE deck delegation despite two sends.
+    assert_eq!(
+        h.transcript.count("(delegating)"),
+        1,
+        "duplicate human msg collapsed"
+    );
+    // Two human-facing notifications: status answer + research-ready.
+    assert_eq!(h.notifications().len(), 2);
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Scenario 4 — back-pressure: promotion budget caps a burst of human asks.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn conversation_promotion_budget_caps_a_burst() {
+    let _g = bus_lock();
+    // The scripted gate intentionally has no promotion budget (that lives in
+    // the real GatePass), so this scenario isolates the *rate limiter*: a
+    // flood of distinct user messages from one source is capped by the token
+    // bucket (capacity 3, no refill).
+    let config = OrchestratorConfig {
+        rate_capacity: 3.0,
+        rate_refill_per_sec: 0.0,
+        ..OrchestratorConfig::default()
+    };
+    let h = Harness::new(config);
+
+    for i in 0..6 {
+        h.ingest(human_msg("slack", "U1", &format!("status check #{i}")));
+    }
+    h.settle().await;
+    h.print("scenario 4: rate-limited burst of human messages");
+
+    // Only 3 of the 6 distinct messages passed the rate limiter → 3 replies.
+    assert_eq!(
+        h.transcript.count("reply  directly"),
+        3,
+        "rate limiter capped the burst"
+    );
+    assert_eq!(h.notifications().len(), 3);
+}

tests/subconscious_fullstack_e2e.rs

@@ -0,0 +1,431 @@
+//! Fully hermetic FULL-STACK e2e: the REAL gate + REAL long-lived session
+//! run against a **mocked LLM provider** — no network, no Ollama, no real
+//! model anywhere (cloud and local provider funnels are both overridden).
+//!
+//! Unlike `subconscious_conversation_e2e.rs` (which injects scripted Gate /
+//! SessionExecutor *above* the model), this test mocks at the *provider*
+//! layer, so the production code paths run for real:
+//!   - `GatePass::evaluate` → `agent::triage::run_triage` → real provider
+//!     funnel → mock → real triage parse → real promote/drop mapping.
+//!   - `LongLivedSession::process_promoted` → `Agent::from_config` (the real
+//!     orchestrator agent + tool loop) → mock → real reserved-thread persistence.
+//!
+//! The mock is installed via the factory's `test_provider_override` seam,
+//! which both provider funnels consult first.
+//!
+//! Gated on the off-by-default `e2e-test-support` feature (the seam is only
+//! compiled then). Run with:
+//! `cargo test --features e2e-test-support --test subconscious_fullstack_e2e -- --nocapture`
+#![cfg(feature = "e2e-test-support")]
+
+use std::sync::{Arc, Mutex as StdMutex, OnceLock};
+
+use async_trait::async_trait;
+
+use openhuman_core::core::event_bus::{init_global, DomainEvent};
+use openhuman_core::openhuman::agent::harness::AgentDefinitionRegistry;
+use openhuman_core::openhuman::config::schema::SubconsciousMode;
+use openhuman_core::openhuman::inference::provider::factory::test_provider_override;
+use openhuman_core::openhuman::inference::provider::traits::{
+    ChatRequest, ChatResponse, ProviderCapabilities, ToolCall,
+};
+use openhuman_core::openhuman::inference::provider::Provider;
+use openhuman_core::openhuman::subconscious::LongLivedSession;
+use openhuman_core::openhuman::subconscious_triggers::{normalize, GatePass};
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Mock LLM provider — deterministic, content-routed, no network.
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Records every prompt the mock saw, and answers deterministically:
+/// - triage turns (recognised by the triage envelope markers) → a JSON
+///   decision (drop for cron, escalate otherwise);
+/// - any other turn (the session/orchestrator agent) → a plain final reply.
+struct MockLlm {
+    seen: StdMutex<Vec<String>>,
+    /// When set, the orchestrator turn emits a real `spawn_subagent` tool
+    /// call so the harness runs an actual sub-agent (native tool mode).
+    spawn: bool,
+}
+
+const SUBAGENT_MARKER: &str = "SUBAGENT_TASK";
+const RESEARCHER_FINDINGS: &str = "Researcher findings: Q3 numbers look healthy.";
+
+impl MockLlm {
+    fn new() -> Arc<Self> {
+        Arc::new(Self {
+            seen: StdMutex::new(Vec::new()),
+            spawn: false,
+        })
+    }
+    fn with_spawning() -> Arc<Self> {
+        Arc::new(Self {
+            seen: StdMutex::new(Vec::new()),
+            spawn: true,
+        })
+    }
+
+    fn triage_decision(&self, joined: &str) -> String {
+        let drop = joined.contains("\"source\": \"cron\"")
+            || joined.contains("\"source\":\"cron\"")
+            || joined.to_lowercase().contains("ignore me");
+        if drop {
+            "{\"action\":\"drop\",\"reason\":\"mock: routine noise\"}".to_string()
+        } else {
+            "{\"action\":\"escalate\",\"target_agent\":\"orchestrator\",\
+              \"prompt\":\"handle this\",\"reason\":\"mock: actionable\"}"
+                .to_string()
+        }
+    }
+
+    fn respond(&self, joined: &str) -> ChatResponse {
+        self.seen.lock().unwrap().push(joined.to_string());
+        let is_triage = joined.contains("DISPLAY_LABEL:") && joined.contains("PAYLOAD:");
+
+        let text = if is_triage {
+            self.triage_decision(joined)
+        } else if joined.contains(RESEARCHER_FINDINGS) {
+            // Orchestrator's follow-up turn after the sub-agent returned →
+            // merge + finish. Checked BEFORE the marker because turn-2's
+            // history echoes the tool-call arguments (which carry the marker).
+            "Mock orchestrator merged the sub-agent findings.".to_string()
+        } else if joined.contains(SUBAGENT_MARKER) {
+            // The spawned researcher sub-agent's own turn → return findings,
+            // no further tool calls (prevents recursive spawning).
+            RESEARCHER_FINDINGS.to_string()
+        } else if self.spawn {
+            // Orchestrator's first turn → delegate via a real spawn_subagent
+            // tool call.
+            return ChatResponse {
+                text: Some("Delegating to the researcher.".to_string()),
+                tool_calls: vec![ToolCall {
+                    id: "call-1".to_string(),
+                    name: "spawn_subagent".to_string(),
+                    arguments: serde_json::json!({
+                        "agent_id": "researcher",
+                        "prompt": format!("{SUBAGENT_MARKER}: investigate the Q3 numbers"),
+                    })
+                    .to_string(),
+                    extra_content: None,
+                }],
+                usage: None,
+                reasoning_content: None,
+            };
+        } else {
+            "Mock orchestrator handled the promoted trigger.".to_string()
+        };
+
+        ChatResponse {
+            text: Some(text),
+            tool_calls: Vec::new(),
+            usage: None,
+            reasoning_content: None,
+        }
+    }
+
+    fn triage_turns(&self) -> usize {
+        self.seen
+            .lock()
+            .unwrap()
+            .iter()
+            .filter(|p| p.contains("DISPLAY_LABEL:") && p.contains("PAYLOAD:"))
+            .count()
+    }
+
+    fn saw(&self, needle: &str) -> bool {
+        self.seen.lock().unwrap().iter().any(|p| p.contains(needle))
+    }
+}
+
+#[async_trait]
+impl Provider for MockLlm {
+    fn capabilities(&self) -> ProviderCapabilities {
+        ProviderCapabilities {
+            // Native tools only when we want to emit a spawn tool call.
+            native_tool_calling: self.spawn,
+            vision: false,
+        }
+    }
+
+    async fn chat_with_system(
+        &self,
+        system_prompt: Option<&str>,
+        message: &str,
+        _model: &str,
+        _temperature: f64,
+    ) -> anyhow::Result<String> {
+        let joined = format!("{}\n{message}", system_prompt.unwrap_or(""));
+        Ok(self.respond(&joined).text.unwrap_or_default())
+    }
+
+    async fn chat(
+        &self,
+        request: ChatRequest<'_>,
+        _model: &str,
+        _temperature: f64,
+    ) -> anyhow::Result<ChatResponse> {
+        let joined = request
+            .messages
+            .iter()
+            .map(|m| format!("{}: {}", m.role, m.content))
+            .collect::<Vec<_>>()
+            .join("\n");
+        Ok(self.respond(&joined))
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Hermetic harness: temp HOME + config + globals + installed mock provider.
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Serializes these tests: they mutate process-global env + install a
+/// process-global provider override.
+fn serial() -> std::sync::MutexGuard<'static, ()> {
+    static LOCK: OnceLock<StdMutex<()>> = OnceLock::new();
+    LOCK.get_or_init(|| StdMutex::new(()))
+        .lock()
+        .unwrap_or_else(|p| p.into_inner())
+}
+
+struct EnvGuard {
+    key: &'static str,
+    old: Option<String>,
+}
+impl EnvGuard {
+    fn set(key: &'static str, val: &str) -> Self {
+        let old = std::env::var(key).ok();
+        std::env::set_var(key, val);
+        Self { key, old }
+    }
+    fn unset(key: &'static str) -> Self {
+        let old = std::env::var(key).ok();
+        std::env::remove_var(key);
+        Self { key, old }
+    }
+}
+impl Drop for EnvGuard {
+    fn drop(&mut self) {
+        match &self.old {
+            Some(v) => std::env::set_var(self.key, v),
+            None => std::env::remove_var(self.key),
+        }
+    }
+}
+
+struct Harness {
+    workspace: std::path::PathBuf,
+    mock: Arc<MockLlm>,
+    _install: test_provider_override::InstallGuard,
+    _home: EnvGuard,
+    _ws: EnvGuard,
+    _keyring: EnvGuard,
+    _tmp: tempfile::TempDir,
+}
+
+fn write_config(openhuman_dir: &std::path::Path) {
+    // api_url is never dialed — the provider override intercepts creation
+    // before any URL is used — but config must parse and disable local AI so
+    // nothing reaches Ollama either.
+    let cfg = r#"api_url = "http://127.0.0.1:9"
+default_model = "mock-model"
+default_temperature = 0.2
+chat_onboarding_completed = true
+
+[secrets]
+encrypt = false
+
+[local_ai]
+enabled = false
+runtime_enabled = false
+"#;
+    let write = |dir: &std::path::Path| {
+        std::fs::create_dir_all(dir).expect("mkdir");
+        std::fs::write(dir.join("config.toml"), cfg).expect("write config");
+    };
+    write(openhuman_dir);
+    write(&openhuman_dir.join("users").join("local"));
+    // Sanity: the config must match the schema.
+    let _: openhuman_core::openhuman::config::Config = toml::from_str(cfg).expect("config schema");
+}
+
+fn harness() -> Harness {
+    harness_with(MockLlm::new())
+}
+
+fn harness_spawning() -> Harness {
+    harness_with(MockLlm::with_spawning())
+}
+
+fn harness_with(mock: Arc<MockLlm>) -> Harness {
+    let tmp = tempfile::tempdir().expect("tempdir");
+    let home = tmp.path().to_path_buf();
+    let openhuman_dir = home.join(".openhuman");
+    write_config(&openhuman_dir);
+    let workspace = home.join("workspace");
+    std::fs::create_dir_all(&workspace).expect("mkdir ws");
+
+    let home_guard = EnvGuard::set("HOME", home.to_str().unwrap());
+    let ws_guard = EnvGuard::set("OPENHUMAN_WORKSPACE", workspace.to_str().unwrap());
+    let keyring_guard = EnvGuard::set("OPENHUMAN_KEYRING_BACKEND", "file");
+
+    // Globals the real pipeline needs.
+    init_global(64);
+    openhuman_core::openhuman::agent::bus::register_agent_handlers();
+    let _ = AgentDefinitionRegistry::init_global_builtins();
+
+    // Install the mock LLM — both provider funnels consult this first.
+    let install = test_provider_override::install(mock.clone());
+
+    Harness {
+        workspace,
+        mock,
+        _install: install,
+        _home: home_guard,
+        _ws: ws_guard,
+        _keyring: keyring_guard,
+        _tmp: tmp,
+    }
+}
+
+fn human_event(message: &str) -> DomainEvent {
+    DomainEvent::ChannelInboundMessage {
+        event_name: "msg".into(),
+        channel: "slack".into(),
+        message: message.into(),
+        sender: Some("U1".into()),
+        reply_target: Some("dm".into()),
+        thread_ts: Some("t1".into()),
+        raw_data: serde_json::Value::Null,
+    }
+}
+
+fn cron_event() -> DomainEvent {
+    DomainEvent::CronJobTriggered {
+        job_id: "nightly".into(),
+        job_name: "nightly recap".into(),
+        job_type: "agent".into(),
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Gate (real triage pipeline) over the mock — promote vs drop.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn fullstack_gate_promotes_human_message_via_real_triage() {
+    let _s = serial();
+    let h = harness();
+    let gate = GatePass::new(100);
+
+    let trigger = normalize(&human_event("can you help with the Q3 plan?"), 1.0).unwrap();
+    let decision = gate.evaluate(&trigger, 1.0).await;
+
+    println!("\n[fullstack] human trigger gate decision: {decision:?}");
+    assert!(
+        decision.is_promote(),
+        "real triage (mock-backed) should promote an actionable human message; got {decision:?}"
+    );
+    assert!(h.mock.triage_turns() >= 1, "the real triage LLM path ran");
+}
+
+#[tokio::test]
+async fn fullstack_gate_drops_cron_noise_via_real_triage() {
+    let _s = serial();
+    let h = harness();
+    let gate = GatePass::new(100);
+
+    let trigger = normalize(&cron_event(), 1.0).unwrap();
+    let decision = gate.evaluate(&trigger, 1.0).await;
+
+    println!("[fullstack] cron trigger gate decision: {decision:?}");
+    assert!(
+        !decision.is_promote(),
+        "routine cron noise should be dropped"
+    );
+    assert!(h.mock.triage_turns() >= 1);
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Session (real orchestrator agent + tool loop) over the mock.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn fullstack_session_runs_real_agent_and_persists() {
+    let _s = serial();
+    let h = harness();
+
+    let session = LongLivedSession::with_thread(
+        h.workspace.clone(),
+        SubconsciousMode::Aggressive,
+        "subconscious:orchestrator".into(),
+    );
+
+    let outcome = session
+        .process_promoted("[user/slack] Please look into the Q3 numbers.", false)
+        .await;
+
+    println!("[fullstack] session outcome: {outcome:?}");
+    let outcome = outcome.expect("real agent turn (mock-backed) should succeed");
+    assert!(
+        outcome.response.contains("Mock orchestrator"),
+        "session returned the mock agent's reply: {}",
+        outcome.response
+    );
+
+    // Real reserved-thread persistence: the user turn + agent reply landed.
+    let msgs = openhuman_core::openhuman::memory_conversations::get_messages(
+        h.workspace.clone(),
+        "subconscious:orchestrator",
+    )
+    .expect("read reserved thread");
+    let senders: Vec<&str> = msgs.iter().map(|m| m.sender.as_str()).collect();
+    assert!(
+        senders.contains(&"user"),
+        "user turn persisted: {senders:?}"
+    );
+    assert!(
+        senders.contains(&"agent"),
+        "agent reply persisted: {senders:?}"
+    );
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Full chain: human → subconscious session → REAL sub-agent → back → human.
+// The mock makes the orchestrator emit a real `spawn_subagent` tool call, so
+// the harness runs an actual researcher sub-agent (inheriting the mock
+// provider), whose output is merged by the orchestrator's follow-up turn.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn fullstack_session_spawns_real_subagent_and_merges() {
+    let _s = serial();
+    let h = harness_spawning();
+
+    let session = LongLivedSession::with_thread(
+        h.workspace.clone(),
+        SubconsciousMode::Aggressive,
+        "subconscious:orchestrator".into(),
+    );
+
+    let outcome = session
+        .process_promoted("[user/slack] Please research the Q3 numbers.", false)
+        .await;
+
+    println!("[fullstack] spawn outcome: {outcome:?}");
+    let outcome = outcome.expect("orchestrator turn with a real sub-agent should succeed");
+
+    // The orchestrator delegated and a REAL researcher sub-agent ran (its turn
+    // carried our SUBAGENT_TASK marker), and its findings flowed back into the
+    // session result — the full human → subconscious → sub-agent → back chain,
+    // all production code, mock-backed model.
+    assert!(
+        h.mock.saw(SUBAGENT_MARKER),
+        "the real harness ran the spawned researcher sub-agent"
+    );
+    assert!(
+        outcome.response.contains("Researcher findings")
+            || outcome.response.contains("merged the sub-agent findings"),
+        "the sub-agent's output flowed back to the session: {}",
+        outcome.response
+    );
+}

tests/subconscious_triggers_e2e.rs

@@ -0,0 +1,592 @@
+//! End-to-end scenario simulation for the subconscious **trigger pipeline**.
+//!
+//! This exercises the real public pipeline functions wired together —
+//! `normalize` → `TriggerRegistry::admit` (dedupe + rate) → gate mapping
+//! (`map_triage_to_gate` + `apply_budget`) → `OrchestratorQueue` — across
+//! every trigger source and gate outcome, plus the real `notify_user`
+//! handoff (event bus + reserved-thread persistence) and the reserved-thread
+//! cold-boot contract the long-lived session depends on.
+//!
+//! It is hermetic (no network, no model) and deterministic: the LLM gate is
+//! simulated by feeding the real `map_triage_to_gate` a `TriageDecision` for
+//! each `TriageAction`, so we test *our* promote/drop/dedupe/budget/queue
+//! logic exhaustively. The actual triage model call and the full agent turn
+//! run through the native bus + global provider and are covered by the
+//! `agent::triage` and agent-harness test suites.
+//!
+//! Run with a visible trace:
+//! `cargo test --test subconscious_triggers_e2e -- --nocapture`
+
+use std::sync::{Arc, Mutex as StdMutex, OnceLock};
+
+use openhuman_core::core::event_bus::{global, init_global, DomainEvent};
+use openhuman_core::openhuman::agent::triage::{TriageAction, TriageDecision};
+use openhuman_core::openhuman::subconscious::{
+    notify_user, ORCHESTRATOR_THREAD_ID, USER_THREAD_ID,
+};
+use openhuman_core::openhuman::subconscious_triggers::gate::{apply_budget, map_triage_to_gate};
+use openhuman_core::openhuman::subconscious_triggers::{
+    normalize, AdmitOutcome, DedupeWindow, EnqueueOutcome, GateDecision, OrchestratorQueue,
+    PromotionBudget, RateLimiter, Trigger, TriggerPriority, TriggerRegistry, TriggerSource,
+};
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Event constructors for the four v1 trigger sources.
+// ─────────────────────────────────────────────────────────────────────────────
+
+fn cron_event(job_id: &str, name: &str) -> DomainEvent {
+    DomainEvent::CronJobTriggered {
+        job_id: job_id.into(),
+        job_name: name.into(),
+        job_type: "agent".into(),
+    }
+}
+
+fn user_event(channel: &str, sender: Option<&str>, message: &str) -> DomainEvent {
+    DomainEvent::ChannelInboundMessage {
+        event_name: "msg".into(),
+        channel: channel.into(),
+        message: message.into(),
+        sender: sender.map(str::to_string),
+        reply_target: Some("dm".into()),
+        thread_ts: Some("t1".into()),
+        raw_data: serde_json::Value::Null,
+    }
+}
+
+fn composio_event(metadata_id: &str, subject: &str) -> DomainEvent {
+    DomainEvent::ComposioTriggerReceived {
+        toolkit: "gmail".into(),
+        trigger: "GMAIL_NEW_GMAIL_MESSAGE".into(),
+        metadata_id: metadata_id.into(),
+        metadata_uuid: format!("{metadata_id}-uuid"),
+        payload: serde_json::json!({ "subject": subject, "body": "PRIVATE BODY" }),
+    }
+}
+
+fn subagent_done_event(task_id: &str, agent_id: &str) -> DomainEvent {
+    DomainEvent::SubagentCompleted {
+        parent_session: "subconscious:orchestrator".into(),
+        task_id: task_id.into(),
+        agent_id: agent_id.into(),
+        elapsed_ms: 10,
+        output_chars: 100,
+        iterations: 2,
+    }
+}
+
+fn triage(action: TriageAction, prompt: Option<&str>) -> TriageDecision {
+    TriageDecision {
+        action,
+        target_agent: prompt.map(|_| "orchestrator".into()),
+        prompt: prompt.map(str::to_string),
+        reason: "simulated gate verdict".into(),
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 1 — normalization: every source maps with the right shape.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn scenario_normalization_covers_all_sources() {
+    let now = 100.0;
+
+    let cron = normalize(&cron_event("j1", "morning brief"), now).expect("cron normalizes");
+    assert_eq!(cron.priority, TriggerPriority::Low);
+    assert!(!cron.external_content);
+    assert!(matches!(cron.source, TriggerSource::Cron { .. }));
+
+    let user = normalize(&user_event("slack", Some("U1"), "what's on my plate?"), now)
+        .expect("user normalizes");
+    assert_eq!(user.priority, TriggerPriority::High);
+    assert!(
+        user.external_content,
+        "inbound channel messages are untrusted"
+    );
+    assert!(user.payload.gate_summary.contains("what's on my plate"));
+
+    let composio =
+        normalize(&composio_event("evt-1", "Invoice #42"), now).expect("composio normalizes");
+    assert_eq!(composio.priority, TriggerPriority::Normal);
+    assert!(composio.external_content, "third-party content is tainted");
+    // Redaction: the private body never reaches the gate summary.
+    assert!(!composio.payload.gate_summary.contains("PRIVATE BODY"));
+    // …but is retained in raw for promotion synthesis.
+    assert_eq!(composio.payload.raw["payload"]["body"], "PRIVATE BODY");
+
+    let subagent =
+        normalize(&subagent_done_event("task-1", "researcher"), now).expect("subagent normalizes");
+    assert!(matches!(
+        subagent.source,
+        TriggerSource::SubagentConclusion { ok: true, .. }
+    ));
+
+    // A self-authored proactive message must NOT become a trigger (anti-loop).
+    assert!(
+        normalize(&user_event("slack", Some("subconscious"), "proactive"), now).is_none(),
+        "orchestrator's own output must not re-trigger it"
+    );
+
+    // Unrelated events are ignored.
+    assert!(normalize(
+        &DomainEvent::ChannelConnected {
+            channel: "slack".into()
+        },
+        now
+    )
+    .is_none());
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 2 — admission: dedupe collapses storms, rate limits floods.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn scenario_dedupe_collapses_webhook_storm() {
+    let reg = TriggerRegistry::with_defaults();
+    // 50 identical Gmail webhooks (same metadata_id) within the TTL window.
+    let mut admitted = 0;
+    let mut duplicates = 0;
+    for i in 0..50 {
+        let ev = composio_event("same-evt", "dup");
+        let t = normalize(&ev, 1000.0 + i as f64 * 0.01).expect("normalize");
+        match reg.admit(&t, 1000.0 + i as f64 * 0.01) {
+            AdmitOutcome::Admitted => admitted += 1,
+            AdmitOutcome::Duplicate => duplicates += 1,
+            AdmitOutcome::RateLimited => {}
+        }
+    }
+    assert_eq!(admitted, 1, "only the first of the storm is admitted");
+    assert_eq!(duplicates, 49);
+}
+
+#[test]
+fn scenario_rate_limit_caps_distinct_events_per_source() {
+    // Tiny bucket: capacity 2, no refill, so the 3rd distinct event of the
+    // same family is rate-limited even though it passes dedupe.
+    let reg = TriggerRegistry::new(DedupeWindow::new(600.0), RateLimiter::new(2.0, 0.0));
+    let outcomes: Vec<AdmitOutcome> = (0..3)
+        .map(|i| {
+            let ev = composio_event(&format!("evt-{i}"), "x");
+            let t = normalize(&ev, 5.0).expect("normalize");
+            reg.admit(&t, 5.0)
+        })
+        .collect();
+    assert_eq!(outcomes[0], AdmitOutcome::Admitted);
+    assert_eq!(outcomes[1], AdmitOutcome::Admitted);
+    assert_eq!(outcomes[2], AdmitOutcome::RateLimited);
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 3 — gate decisions: every triage action maps correctly.
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn scenario_gate_maps_every_triage_action() {
+    let now = 0.0;
+    let budget = PromotionBudget::new(100);
+    let base = normalize(&composio_event("g1", "subj"), now).expect("normalize");
+
+    // Drop → dropped, not acknowledged.
+    let d = apply_budget(
+        map_triage_to_gate(&triage(TriageAction::Drop, None), &base),
+        &budget,
+        now,
+    );
+    assert!(matches!(
+        d,
+        GateDecision::Drop {
+            acknowledge: false,
+            ..
+        }
+    ));
+
+    // Acknowledge → dropped but acknowledged.
+    let a = apply_budget(
+        map_triage_to_gate(&triage(TriageAction::Acknowledge, None), &base),
+        &budget,
+        now,
+    );
+    assert!(matches!(
+        a,
+        GateDecision::Drop {
+            acknowledge: true,
+            ..
+        }
+    ));
+
+    // React → promote, keeping the trigger's own priority.
+    let r = map_triage_to_gate(&triage(TriageAction::React, Some("ack it")), &base);
+    match r {
+        GateDecision::Promote {
+            priority,
+            synthesized_summary,
+            ..
+        } => {
+            assert_eq!(priority, base.priority);
+            assert!(synthesized_summary.contains("ack it"));
+        }
+        other => panic!("expected promote, got {other:?}"),
+    }
+
+    // Escalate → promote at >= High.
+    let e = map_triage_to_gate(&triage(TriageAction::Escalate, Some("draft reply")), &base);
+    match e {
+        GateDecision::Promote { priority, .. } => assert!(priority >= TriggerPriority::High),
+        other => panic!("expected promote, got {other:?}"),
+    }
+}
+
+#[test]
+fn scenario_promotion_budget_exhaustion_downgrades_to_ack_drop() {
+    let now = 0.0;
+    let budget = PromotionBudget::new(1); // one promotion per hour
+    let base = normalize(&user_event("slack", Some("U1"), "urgent!"), now).expect("normalize");
+    let escalate = || map_triage_to_gate(&triage(TriageAction::Escalate, Some("do it")), &base);
+
+    // First escalation promotes.
+    assert!(apply_budget(escalate(), &budget, now).is_promote());
+    // Second within the hour is downgraded to an acknowledged drop — noted,
+    // but no reasoning-tier session run is spent.
+    match apply_budget(escalate(), &budget, now) {
+        GateDecision::Drop {
+            acknowledge,
+            reason,
+        } => {
+            assert!(acknowledge);
+            assert!(reason.contains("budget exhausted"));
+        }
+        other => panic!("expected budget downgrade, got {other:?}"),
+    }
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 4 — queue: priority ordering + overflow eviction.
+// ─────────────────────────────────────────────────────────────────────────────
+
+fn promoted(label: &str, priority: TriggerPriority) -> Trigger {
+    let mut t = normalize(&cron_event("j", label), 0.0).expect("normalize");
+    t.display_label = label.into();
+    t.priority = priority;
+    t
+}
+
+#[test]
+fn scenario_queue_serves_highest_priority_first() {
+    let q = OrchestratorQueue::new(16);
+    q.push(promoted("cron-low", TriggerPriority::Low));
+    q.push(promoted("user-high", TriggerPriority::High));
+    q.push(promoted("webhook-normal", TriggerPriority::Normal));
+    q.push(promoted("interrupt-urgent", TriggerPriority::Urgent));
+
+    let drained: Vec<String> = std::iter::from_fn(|| q.pop())
+        .map(|t| t.display_label)
+        .collect();
+    assert_eq!(
+        drained,
+        vec![
+            "interrupt-urgent",
+            "user-high",
+            "webhook-normal",
+            "cron-low"
+        ]
+    );
+}
+
+#[test]
+fn scenario_queue_overflow_sheds_lowest_priority() {
+    let q = OrchestratorQueue::new(2);
+    assert_eq!(
+        q.push(promoted("low", TriggerPriority::Low)),
+        EnqueueOutcome::Accepted
+    );
+    assert_eq!(
+        q.push(promoted("normal", TriggerPriority::Normal)),
+        EnqueueOutcome::Accepted
+    );
+    // Full; an Urgent arrival evicts the lowest-priority held item.
+    match q.push(promoted("urgent", TriggerPriority::Urgent)) {
+        EnqueueOutcome::EvictedLowest { evicted } => assert_eq!(evicted.display_label, "low"),
+        other => panic!("expected eviction, got {other:?}"),
+    }
+    // Full again; a Low arrival is dropped rather than evicting a better item.
+    assert_eq!(
+        q.push(promoted("late-low", TriggerPriority::Low)),
+        EnqueueOutcome::DroppedIncoming
+    );
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 5 — full pipeline simulation across scenarios, with a printed trace.
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// One scenario: a raw event + the gate verdict the (simulated) LLM returns.
+struct Scenario {
+    name: &'static str,
+    event: DomainEvent,
+    action: TriageAction,
+    prompt: Option<&'static str>,
+}
+
+/// Outcome of running a scenario through the whole pipeline.
+#[derive(Debug)]
+enum Outcome {
+    Ignored,
+    Duplicate,
+    RateLimited,
+    Dropped,
+    Promoted { priority: TriggerPriority },
+}
+
+/// Drive one event through normalize → admit → gate → enqueue, returning what
+/// happened and (on promotion) pushing onto `queue`.
+fn run_scenario(
+    s: &Scenario,
+    reg: &TriggerRegistry,
+    budget: &PromotionBudget,
+    queue: &OrchestratorQueue,
+    now: f64,
+) -> Outcome {
+    let Some(trigger) = normalize(&s.event, now) else {
+        return Outcome::Ignored;
+    };
+    match reg.admit(&trigger, now) {
+        AdmitOutcome::Duplicate => return Outcome::Duplicate,
+        AdmitOutcome::RateLimited => return Outcome::RateLimited,
+        AdmitOutcome::Admitted => {}
+    }
+    let decision = apply_budget(
+        map_triage_to_gate(&triage(s.action, s.prompt), &trigger),
+        budget,
+        now,
+    );
+    match decision {
+        GateDecision::Drop { .. } => Outcome::Dropped,
+        GateDecision::Promote {
+            priority,
+            synthesized_summary,
+            ..
+        } => {
+            let mut item = trigger;
+            item.priority = priority;
+            item.payload.gate_summary = synthesized_summary;
+            queue.push(item);
+            Outcome::Promoted { priority }
+        }
+    }
+}
+
+#[test]
+fn scenario_full_pipeline_simulation_with_trace() {
+    let reg = TriggerRegistry::with_defaults();
+    let budget = PromotionBudget::new(2); // tight budget to exercise exhaustion
+    let queue = OrchestratorQueue::new(64);
+
+    let scenarios = vec![
+        Scenario {
+            name: "cron tick — gate drops routine noise",
+            event: cron_event("nightly", "nightly recap"),
+            action: TriageAction::Drop,
+            prompt: None,
+        },
+        Scenario {
+            name: "user message — gate escalates (promote #1)",
+            event: user_event("slack", Some("U1"), "can you prep the Q3 deck?"),
+            action: TriageAction::Escalate,
+            prompt: Some("prepare the Q3 deck"),
+        },
+        Scenario {
+            name: "composio gmail — gate reacts (promote #2)",
+            event: composio_event("inv-1", "Invoice overdue"),
+            action: TriageAction::React,
+            prompt: Some("flag the overdue invoice"),
+        },
+        Scenario {
+            name: "composio gmail DUPLICATE — collapsed by dedupe",
+            event: composio_event("inv-1", "Invoice overdue"),
+            action: TriageAction::React,
+            prompt: Some("flag the overdue invoice"),
+        },
+        Scenario {
+            name: "subagent conclusion — would promote but budget exhausted",
+            event: subagent_done_event("task-9", "researcher"),
+            action: TriageAction::Escalate,
+            prompt: Some("merge the research findings"),
+        },
+        Scenario {
+            name: "self-authored proactive echo — ignored (anti-loop)",
+            event: user_event("slack", Some("subconscious"), "FYI: deck is ready"),
+            action: TriageAction::Escalate,
+            prompt: Some("should never run"),
+        },
+    ];
+
+    println!("\n=== subconscious trigger pipeline — scenario trace ===");
+    let mut promotions = 0;
+    let mut trace = Vec::new();
+    for (i, s) in scenarios.iter().enumerate() {
+        let outcome = run_scenario(s, &reg, &budget, &queue, 1000.0 + i as f64);
+        if matches!(outcome, Outcome::Promoted { .. }) {
+            promotions += 1;
+        }
+        println!("  [{i}] {:<55} -> {outcome:?}", s.name);
+        trace.push(outcome);
+    }
+    println!("  queue depth after gating: {}", queue.len());
+    println!("=== drain (highest priority first) ===");
+    while let Some(item) = queue.pop() {
+        println!(
+            "  run -> {:<24} priority={}",
+            item.display_label,
+            item.priority.as_str()
+        );
+    }
+
+    // Assertions on what the pipeline decided.
+    assert!(matches!(trace[0], Outcome::Dropped), "cron noise dropped");
+    assert!(
+        matches!(trace[1], Outcome::Promoted { priority } if priority >= TriggerPriority::High),
+        "user escalation promoted at >= High"
+    );
+    assert!(
+        matches!(trace[2], Outcome::Promoted { .. }),
+        "gmail react promoted"
+    );
+    assert!(
+        matches!(trace[3], Outcome::Duplicate),
+        "duplicate gmail collapsed"
+    );
+    assert!(
+        matches!(trace[4], Outcome::Dropped),
+        "budget (2) exhausted → 3rd promotion downgraded to drop"
+    );
+    assert!(matches!(trace[5], Outcome::Ignored), "self-echo ignored");
+    assert_eq!(promotions, 2, "exactly two promotions within the budget");
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 6 — notify_user: real event bus + reserved user-thread persistence.
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Serializes tests that touch the process-global event bus so concurrent
+/// captures don't cross-talk.
+fn bus_lock() -> std::sync::MutexGuard<'static, ()> {
+    static LOCK: OnceLock<StdMutex<()>> = OnceLock::new();
+    LOCK.get_or_init(|| StdMutex::new(()))
+        .lock()
+        .unwrap_or_else(|p| p.into_inner())
+}
+
+#[tokio::test]
+async fn scenario_notify_user_delivers_and_persists() {
+    let _guard = bus_lock();
+    init_global(64);
+
+    let captured: Arc<StdMutex<Vec<DomainEvent>>> = Arc::new(StdMutex::new(Vec::new()));
+    let sink = Arc::clone(&captured);
+    let _sub = global()
+        .expect("bus initialized")
+        .on("e2e-notify-capture", move |event| {
+            let sink = Arc::clone(&sink);
+            let event = event.clone();
+            Box::pin(async move {
+                if let DomainEvent::ProactiveMessageRequested { .. } = &event {
+                    sink.lock().unwrap().push(event);
+                }
+            })
+        });
+
+    let tmp = tempfile::tempdir().expect("tempdir");
+    let workspace = tmp.path().to_path_buf();
+
+    let unique = "E2E-NOTIFY-MARKER-7321";
+    notify_user(workspace.clone(), unique, Some("heads-up"));
+
+    // Give the async broadcast a moment to deliver.
+    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+
+    // 1) The proactive delivery event fired, tagged as subconscious-sourced.
+    let events = captured.lock().unwrap();
+    let found = events.iter().any(|e| match e {
+        DomainEvent::ProactiveMessageRequested {
+            source, message, ..
+        } => source == "subconscious" && message == unique,
+        _ => false,
+    });
+    assert!(
+        found,
+        "notify_user must publish a ProactiveMessageRequested event"
+    );
+
+    // 2) The message landed in the reserved user-facing thread.
+    let persisted =
+        openhuman_core::openhuman::memory_conversations::get_messages(workspace, USER_THREAD_ID)
+            .expect("read user thread");
+    assert!(
+        persisted
+            .iter()
+            .any(|m| m.content == unique && m.sender == "agent"),
+        "notify_user must persist to the user-facing thread"
+    );
+}
+
+// ─────────────────────────────────────────────────────────────────────────────
+// Section 7 — reserved-thread cold-boot contract (what the session resumes from).
+// ─────────────────────────────────────────────────────────────────────────────
+
+#[test]
+fn scenario_reserved_threads_are_distinct_and_persist() {
+    use openhuman_core::openhuman::memory_conversations::{
+        append_message, ensure_thread, get_messages, ConversationMessage, CreateConversationThread,
+    };
+
+    assert_ne!(
+        ORCHESTRATOR_THREAD_ID, USER_THREAD_ID,
+        "orchestrator and user threads must be distinct"
+    );
+
+    let tmp = tempfile::tempdir().expect("tempdir");
+    let workspace = tmp.path().to_path_buf();
+
+    // The reserved thread must exist before appending (the production code
+    // ensures this lazily; mirror it here).
+    ensure_thread(
+        workspace.clone(),
+        CreateConversationThread {
+            id: ORCHESTRATOR_THREAD_ID.into(),
+            title: "Subconscious Orchestrator".into(),
+            created_at: "2026-06-12T00:00:00Z".into(),
+            parent_thread_id: None,
+            labels: None,
+            personality_id: None,
+        },
+    )
+    .expect("ensure thread");
+
+    // Simulate prior orchestrator history that a long-lived session would
+    // cold-boot resume from via seed_resume_from_messages.
+    for (sender, content) in [
+        ("user", "[composio/gmail] new invoice"),
+        ("agent", "Noted; I'll watch for the follow-up."),
+    ] {
+        append_message(
+            workspace.clone(),
+            ORCHESTRATOR_THREAD_ID,
+            ConversationMessage {
+                id: format!("m-{sender}"),
+                content: content.into(),
+                message_type: "text".into(),
+                extra_metadata: serde_json::Value::Null,
+                sender: sender.into(),
+                created_at: "2026-06-12T00:00:00Z".into(),
+            },
+        )
+        .expect("append");
+    }
+
+    let history = get_messages(workspace, ORCHESTRATOR_THREAD_ID).expect("read");
+    assert_eq!(history.len(), 2);
+    assert_eq!(history[0].sender, "user");
+    assert_eq!(history[1].sender, "agent");
+}