diff --git a/src/core/memory_cli.rs b/src/core/memory_cli.rs index 207d558fb..32d89b90e 100644 --- a/src/core/memory_cli.rs +++ b/src/core/memory_cli.rs @@ -132,6 +132,7 @@ fn run_ingest(args: &[String]) -> Result<()> { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }; let ingestion_config = MemoryIngestionConfig::default(); diff --git a/src/openhuman/agent/bus.rs b/src/openhuman/agent/bus.rs index f0a3542b0..0ee18b1fc 100644 --- a/src/openhuman/agent/bus.rs +++ b/src/openhuman/agent/bus.rs @@ -20,6 +20,7 @@ use tokio::sync::mpsc; use crate::core::event_bus::register_native_global; use crate::openhuman::agent::progress::AgentProgress; +use crate::openhuman::agent::turn_origin::{self, AgentTurnOrigin}; use crate::openhuman::config::MultimodalConfig; use crate::openhuman::inference::provider::{ChatMessage, Provider}; use crate::openhuman::prompt_injection::{ @@ -33,12 +34,6 @@ use super::harness::{run_tool_call_loop, with_current_sandbox_mode}; /// Method name used to dispatch an agentic turn through the native bus. pub const AGENT_RUN_TURN_METHOD: &str = "agent.run_turn"; -/// Full owned payload for a single agentic turn executed through the bus. -/// -/// All fields are either owned values, [`Arc`]s, or channel handles — the -/// bus carries them by value without touching serialization. Consumers can -/// therefore pass trait objects (`Arc`, tool trait-object -/// registries) and streaming senders (`on_delta`) through unchanged. /// Full owned payload for a single agentic turn executed through the bus. /// /// All fields are either owned values, [`Arc`]s, or channel handles — the @@ -128,6 +123,20 @@ pub struct AgentTurnRequest { /// progressively edit outbound messages. `None` disables streaming /// status updates for this turn. pub on_progress: Option>, + + /// Trust/routing label for this turn. The bus handler scopes + /// [`AGENT_TURN_ORIGIN`](crate::openhuman::agent::turn_origin::AGENT_TURN_ORIGIN) + /// around the tool loop so the approval gate (and any other + /// origin-aware policy) sees the same value the caller intended. + /// + /// Native-bus dispatch crosses a `tokio::spawn` boundary inside the + /// registry, so callers cannot rely on `AGENT_TURN_ORIGIN` propagating + /// implicitly — the origin must travel as an owned field. + /// + /// Defaults to [`AgentTurnOrigin::Unknown`] so any caller that fails + /// to populate this fails closed at the gate rather than silently + /// inheriting trusted-automation semantics. + pub origin: AgentTurnOrigin, } /// Final response from an agentic turn. @@ -163,6 +172,7 @@ pub fn register_agent_handlers() { visible_tool_names, extra_tools, on_progress, + origin, } = req; tracing::debug!( @@ -239,35 +249,45 @@ pub fn register_agent_handlers() { .map(|def| def.sandbox_mode) .unwrap_or(SandboxMode::None); - let text = with_current_sandbox_mode(sandbox_mode, async { - run_tool_call_loop( - provider.as_ref(), - &mut history, - tools_registry.as_ref(), - &provider_name, - &model, - temperature, - silent, - &channel_name, - &multimodal, - &multimodal_files, - max_tool_iterations, - on_delta, - visible_tool_names.as_ref(), - &extra_tools, - on_progress, - // Bus path runs ad-hoc agent turns without an Agent - // handle, so we pass None — payload summarization is - // wired into the orchestrator session via Agent::turn, - // not the bus dispatcher. - None, - // Use the default (allow-all) tool policy. Custom - // policies can be wired in via AgentTurnRequest when - // per-channel policy configuration is added (#2134). - &crate::openhuman::tools::policy::DefaultToolPolicy, - ) - .await - }) + // Scope the caller-supplied origin around the tool loop so + // the approval gate (and any other origin-aware policy) sees + // the same trust label the entry point intended. Native-bus + // dispatch crosses a `tokio::spawn` boundary inside the + // registry, so re-scoping here is mandatory — the + // task-local does NOT propagate across that boundary + // implicitly. + let text = turn_origin::with_origin( + origin, + with_current_sandbox_mode(sandbox_mode, async { + run_tool_call_loop( + provider.as_ref(), + &mut history, + tools_registry.as_ref(), + &provider_name, + &model, + temperature, + silent, + &channel_name, + &multimodal, + &multimodal_files, + max_tool_iterations, + on_delta, + visible_tool_names.as_ref(), + &extra_tools, + on_progress, + // Bus path runs ad-hoc agent turns without an Agent + // handle, so we pass None — payload summarization is + // wired into the orchestrator session via Agent::turn, + // not the bus dispatcher. + None, + // Use the default (allow-all) tool policy. Custom + // policies can be wired in via AgentTurnRequest when + // per-channel policy configuration is added (#2134). + &crate::openhuman::tools::policy::DefaultToolPolicy, + ) + .await + }), + ) .await .map_err(|e| e.to_string())?; @@ -415,6 +435,7 @@ mod tests { visible_tool_names: None, extra_tools: Vec::new(), on_progress: None, + origin: AgentTurnOrigin::Cli, } } diff --git a/src/openhuman/agent/harness/memory_context.rs b/src/openhuman/agent/harness/memory_context.rs index a5ca26e25..dc2ac329e 100644 --- a/src/openhuman/agent/harness/memory_context.rs +++ b/src/openhuman/agent/harness/memory_context.rs @@ -279,6 +279,7 @@ mod tests { timestamp: "now".into(), session_id: None, score, + taint: Default::default(), } } @@ -297,6 +298,7 @@ mod tests { timestamp: "now".into(), session_id: Some(session_id.into()), score, + taint: Default::default(), } } diff --git a/src/openhuman/agent/harness/memory_context_safety.rs b/src/openhuman/agent/harness/memory_context_safety.rs index 46e61a720..ca823fcb1 100644 --- a/src/openhuman/agent/harness/memory_context_safety.rs +++ b/src/openhuman/agent/harness/memory_context_safety.rs @@ -146,6 +146,7 @@ mod tests { timestamp: "2026-05-20T00:00:00Z".into(), session_id: None, score: None, + taint: Default::default(), } } diff --git a/src/openhuman/agent/memory_loader.rs b/src/openhuman/agent/memory_loader.rs index a8d6454ca..3cc6319d8 100644 --- a/src/openhuman/agent/memory_loader.rs +++ b/src/openhuman/agent/memory_loader.rs @@ -543,6 +543,7 @@ mod tests { timestamp: "2026-04-22T00:00:00Z".to_string(), session_id: None, score, + taint: Default::default(), } } @@ -570,6 +571,7 @@ mod tests { timestamp: "2026-05-25T00:00:00Z".into(), session_id: None, score: None, + taint: Default::default(), }]); let out = DefaultMemoryLoader::default() @@ -600,6 +602,7 @@ mod tests { timestamp: "2026-04-22T00:00:00Z".into(), session_id: Some("thr_old".into()), score: Some(0.9), + taint: Default::default(), }]); let out = DefaultMemoryLoader::default() @@ -632,6 +635,7 @@ mod tests { timestamp: "2026-04-22T00:00:00Z".into(), session_id: Some("thr_old".into()), score: Some(0.9), + taint: Default::default(), }, MemoryEntry { id: "id-2".into(), @@ -642,6 +646,7 @@ mod tests { timestamp: "2026-04-22T00:00:00Z".into(), session_id: None, score: Some(0.9), + taint: Default::default(), }, ]); @@ -700,6 +705,7 @@ mod tests { timestamp: "2026-05-15T00:00:00Z".into(), session_id: Some(session_id.into()), score, + taint: Default::default(), } } diff --git a/src/openhuman/agent/mod.rs b/src/openhuman/agent/mod.rs index ce662d84c..99bbe84fa 100644 --- a/src/openhuman/agent/mod.rs +++ b/src/openhuman/agent/mod.rs @@ -46,6 +46,12 @@ pub mod tool_policy; pub mod tools; pub mod tree_loader; pub mod triage; +/// Turn-origin task-local — explicit trust/routing label scoped by every +/// entry point that invokes the agent (web chat, channel runtime, +/// subconscious, cron, CLI). Read by the approval gate to make +/// origin-aware decisions rather than inferring trust from the absence of +/// `APPROVAL_CHAT_CONTEXT`. +pub mod turn_origin; pub use schemas::{ all_controller_schemas as all_agent_controller_schemas, all_registered_controllers as all_agent_registered_controllers, diff --git a/src/openhuman/agent/task_dispatcher.rs b/src/openhuman/agent/task_dispatcher.rs index fd8e6a27d..30a241f9d 100644 --- a/src/openhuman/agent/task_dispatcher.rs +++ b/src/openhuman/agent/task_dispatcher.rs @@ -378,9 +378,16 @@ async fn run_autonomous( run_id.get(..8).unwrap_or(run_id) )); - with_autonomous_iter_cap(TASK_RUN_MAX_ITERATIONS, agent.run_single(prompt)) - .await - .map_err(|e| format!("{e:#}")) + // Sub-agent task runs are internal to the agent harness — the user + // already authorized the parent turn that dispatched this task. Label + // as CLI so the approval gate doesn't fail closed on internal + // sub-agent invocations. + crate::openhuman::agent::turn_origin::with_origin( + crate::openhuman::agent::turn_origin::AgentTurnOrigin::Cli, + with_autonomous_iter_cap(TASK_RUN_MAX_ITERATIONS, agent.run_single(prompt)), + ) + .await + .map_err(|e| format!("{e:#}")) } /// Deterministic board write-back: the dispatcher owns the card lifecycle. diff --git a/src/openhuman/agent/triage/evaluator.rs b/src/openhuman/agent/triage/evaluator.rs index b996c4177..390c4e4db 100644 --- a/src/openhuman/agent/triage/evaluator.rs +++ b/src/openhuman/agent/triage/evaluator.rs @@ -490,6 +490,16 @@ async fn try_arm( visible_tool_names: None, extra_tools: Vec::new(), on_progress: None, + // Triage processes untrusted inbound channel text. Label it as + // ExternalChannel so the approval gate treats any external_effect + // tool call originating from this turn as remote-attacker input + // (the triage agent doesn't usually invoke such tools — it + // classifies and routes — but label correctly for defense in depth). + origin: crate::openhuman::agent::turn_origin::AgentTurnOrigin::ExternalChannel { + channel: envelope.source.slug().to_string(), + reply_target: envelope.display_label.clone(), + message_id: envelope.external_id.clone(), + }, }; let response = match request_native_global::( diff --git a/src/openhuman/agent/turn_origin.rs b/src/openhuman/agent/turn_origin.rs new file mode 100644 index 000000000..6caec6186 --- /dev/null +++ b/src/openhuman/agent/turn_origin.rs @@ -0,0 +1,153 @@ +//! Agent turn origin — the trust/routing label attached to every agent +//! `run_turn` invocation. Read by [`crate::openhuman::approval::ApprovalGate`] +//! and [`crate::openhuman::agent_tool_policy::ToolPolicyEngine`] to make +//! consistent decisions across web, channel, subconscious, and cron entry +//! points without relying on the *absence* of other task-locals as a signal. +//! +//! Every entry point that drives the agent loop ([`crate::openhuman::channels::providers::web`], +//! [`crate::openhuman::channels::runtime::dispatch`], [`crate::openhuman::subconscious`], +//! [`crate::openhuman::cron`], CLI) MUST scope a real [`AgentTurnOrigin`] +//! around its `run_turn` invocation. Any path that fails to do so is treated +//! as [`AgentTurnOrigin::Unknown`] by the gate and the call fails closed. + +/// Identifies who scheduled the current agent turn so the approval gate can +/// pick the correct policy: surface to the user, persist for an +/// out-of-band approval surface, run trusted-automation through, or fail +/// closed. +/// +/// This is a typed task-local label, not a credential — it is set by the +/// entry point that owns the turn and read by [`crate::openhuman::approval`] +/// alongside the existing per-turn chat context. +#[derive(Clone, Debug)] +pub enum AgentTurnOrigin { + /// Live user chat in the desktop / web UI. The existing + /// [`crate::openhuman::approval::ApprovalChatContext`] task-local is + /// scoped alongside this so the approval gate has a thread / client to + /// route the prompt back to. + WebChat { + thread_id: String, + client_id: String, + }, + /// Inbound message from a non-web channel (Telegram / Discord / Slack / + /// Yuanbao / etc.). External-effect tools must persist a + /// `pending_approvals` row for the audit trail; the parked future will + /// TTL-deny because no caller picks up the chat-routed approval on this + /// surface yet — which is the correct fail-closed default for remote + /// inputs. + ExternalChannel { + channel: String, + reply_target: String, + message_id: String, + }, + /// Internal automation the user explicitly authorized (cron job the + /// user created, subconscious tick on internal-only memory). `source` + /// carries enough info for the gate to apply the right per-source + /// allowlist. + TrustedAutomation { + job_id: String, + source: TrustedAutomationSource, + }, + /// Command-line / sub-agent / one-off internal invocation. + Cli, + /// Unlabelled — gate fails closed. Every entry point MUST scope a real + /// origin before invoking the agent. + Unknown, +} + +/// Sub-classification for [`AgentTurnOrigin::TrustedAutomation`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TrustedAutomationSource { + /// Cron job created and authorized by the user. + Cron, + /// Subconscious tick whose memory context is internal-only. + Subconscious, + /// Subconscious tick whose memory context includes chunks ingested + /// from an external sync source (Gmail / Slack / Notion / etc.). + /// Treated as untrusted: external-effect tool surface blocked. + SubconsciousTainted, +} + +tokio::task_local! { + /// Per-turn agent origin. Scoped by entry points (web channel, channel + /// runtime dispatch, subconscious loop, cron scheduler, CLI) around the + /// `run_turn` invocation. Read by the approval gate to make + /// origin-aware decisions. + pub static AGENT_TURN_ORIGIN: AgentTurnOrigin; +} + +/// Scope `origin` for the duration of `fut`. Mirrors the existing +/// [`crate::openhuman::approval::APPROVAL_CHAT_CONTEXT`] scope pattern. +/// +/// The inner future is `Box::pin`-ed before being handed to the task-local +/// scope so the combined `with_origin(... scope(... run_turn(...)))` future +/// state machine stays heap-allocated. The agent loop downstream of this +/// scope can be deep (tool dispatch, recursive sub-agent invocations, LLM +/// streaming), and stacking two task-local scopes plus the agent loop on a +/// 2 MiB worker stack reliably blows the test runtime — same shape as the +/// fix in PR #3151. Box-pinning here is the single-point remediation that +/// covers every caller (web channel, channel runtime, subconscious, cron, +/// CLI). +pub async fn with_origin(origin: AgentTurnOrigin, fut: F) -> F::Output { + AGENT_TURN_ORIGIN.scope(origin, Box::pin(fut)).await +} + +/// Try to read the current origin. Returns `None` when no caller scoped one +/// (legacy callers that haven't been migrated yet — the gate maps this to +/// [`AgentTurnOrigin::Unknown`] / fail-closed). +pub fn current() -> Option { + AGENT_TURN_ORIGIN.try_with(|o| o.clone()).ok() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn with_origin_scopes_correctly_and_unscopes_on_exit() { + // Outside any scope: current() returns None. + assert!(current().is_none()); + + let observed = with_origin(AgentTurnOrigin::Cli, async { + // Inside the scope: current() returns the scoped origin. + current() + }) + .await; + assert!(matches!(observed, Some(AgentTurnOrigin::Cli))); + + // After the scope exits, current() is None again. + assert!(current().is_none()); + } + + #[tokio::test] + async fn current_returns_none_outside_scope() { + assert!(current().is_none()); + } + + #[tokio::test] + async fn current_returns_inner_origin_on_nested_scope() { + let observed = with_origin( + AgentTurnOrigin::WebChat { + thread_id: "outer".into(), + client_id: "c-outer".into(), + }, + async { + with_origin( + AgentTurnOrigin::TrustedAutomation { + job_id: "j-1".into(), + source: TrustedAutomationSource::Cron, + }, + async { current() }, + ) + .await + }, + ) + .await; + match observed { + Some(AgentTurnOrigin::TrustedAutomation { job_id, source }) => { + assert_eq!(job_id, "j-1"); + assert_eq!(source, TrustedAutomationSource::Cron); + } + other => panic!("expected inner TrustedAutomation, got {other:?}"), + } + } +} diff --git a/src/openhuman/agent_tool_policy/engine.rs b/src/openhuman/agent_tool_policy/engine.rs index f11ecdba9..c540be853 100644 --- a/src/openhuman/agent_tool_policy/engine.rs +++ b/src/openhuman/agent_tool_policy/engine.rs @@ -106,9 +106,18 @@ fn permission_for_channel( channel: &str, ) -> PermissionLevel { if channel_permissions.is_empty() { + // Empty map means "operator hasn't configured a per-channel + // policy yet" — keep the legacy unrestricted surface so existing + // installs (and unit fixtures that don't seed the map) keep + // working. The hardening lands at the config layer: + // [`AgentConfig::migrate_channel_permissions_if_legacy`] runs at + // startup on legacy installs and seeds the map with safe + // per-channel defaults so the cap actually engages on the very + // first boot after upgrade. Once any entry exists, unknown + // channels fall back to ReadOnly (the `None` arm below). log::debug!( target: "openhuman::agent_tool_policy", - "[tool-policy] channel permissions empty; preserving legacy unrestricted surface channel={}", + "[tool-policy] channel permissions empty; preserving legacy unrestricted surface channel={} (config migration seeds per-channel defaults on first boot)", channel ); return PermissionLevel::Dangerous; @@ -243,6 +252,13 @@ mod tests { #[test] fn empty_channel_config_preserves_legacy_full_tool_surface() { + // Empty channel_permissions preserves the legacy unrestricted + // tool surface (channel cap returns Dangerous). The real-world + // hardening landed at the config layer: legacy installs are + // migrated via `AgentConfig::migrate_channel_permissions_if_legacy` + // on first boot, which seeds per-channel defaults so the cap + // actually engages. Tests that don't exercise that migration + // path keep their legacy behavior. let session = ToolPolicyEngine::build_session( "orchestrator", "web", diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index 6fe2ff5ff..b00439cfd 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -38,6 +38,7 @@ use parking_lot::Mutex; use tokio::sync::oneshot; use crate::core::event_bus::{publish_global, DomainEvent}; +use crate::openhuman::agent::turn_origin::{self, AgentTurnOrigin, TrustedAutomationSource}; use crate::openhuman::config::Config; use crate::openhuman::security::POLICY_DENIED_MARKER; @@ -210,24 +211,116 @@ impl ApprovalGate { return (GateOutcome::Allow, None); } + // Origin tells us who scheduled this turn. Entry points (web channel, + // channel runtime, subconscious, cron, CLI) scope a typed + // `AgentTurnOrigin` around `run_turn`. Unlabelled callers map to + // `Unknown`, which is denied — the gate refuses to execute an + // external_effect tool from an unlabelled call site. + let origin = turn_origin::current().unwrap_or(AgentTurnOrigin::Unknown); + // Chat context (thread/client id) for routing the yes/no reply — set by // the web channel around the agent run; absent for non-chat callers. let chat_ctx = APPROVAL_CHAT_CONTEXT.try_with(|c| c.clone()).ok(); let chat_thread_id = chat_ctx.as_ref().map(|c| c.thread_id.clone()); let chat_client_id = chat_ctx.as_ref().map(|c| c.client_id.clone()); - // The gate is interactive: it only engages when there's a live chat turn - // to surface the prompt to and a human to answer it. Background / triage - // / cron turns carry no `ApprovalChatContext` — they are pre-authorized - // autonomous automation, and gating them would park with nobody to - // answer (→ TTL timeout → deny), stalling the automation. So with no - // chat context, allow the call straight through. - if chat_ctx.is_none() { - tracing::debug!( - tool = tool_name, - "[approval::gate] no chat context (non-interactive turn) — not gating" - ); - return (GateOutcome::Allow, None); + // Branch by origin. Web chat parks for an in-app approval; external + // channel persists an audit row and TTL-denies (no routable approval + // surface yet); trusted automation (cron, internal-only subconscious) + // is allowed through unchanged; tainted subconscious — a tick whose + // memory context contains external-sync chunks — is denied because + // remote text could otherwise steer it into an external_effect tool; + // CLI keeps the legacy allow; Unknown fails closed. + match &origin { + AgentTurnOrigin::WebChat { .. } => { + // Fall through to the existing chat-routed parking flow below. + } + AgentTurnOrigin::ExternalChannel { + channel, + reply_target, + message_id, + } => { + tracing::info!( + tool = tool_name, + channel = %channel, + reply_target = %reply_target, + message_id = %message_id, + "[approval::gate] external channel turn — persisting audit row and parking \ + (will TTL-deny until a routable channel approval surface ships)" + ); + // Fall through to the parking flow: a `pending_approvals` row + // is persisted (audit trail) and the future TTL-denies. We do + // NOT short-circuit to Allow here — remote inputs are + // untrusted, and there is no UI surface to route a yes/no on + // a non-web channel right now. + } + AgentTurnOrigin::TrustedAutomation { + source: TrustedAutomationSource::Cron, + job_id, + } => { + tracing::debug!( + tool = tool_name, + job_id = %job_id, + "[approval::gate] trusted cron automation — allowing without prompt" + ); + return (GateOutcome::Allow, None); + } + AgentTurnOrigin::TrustedAutomation { + source: TrustedAutomationSource::Subconscious, + job_id, + } => { + tracing::debug!( + tool = tool_name, + job_id = %job_id, + "[approval::gate] trusted internal subconscious tick — allowing without prompt" + ); + return (GateOutcome::Allow, None); + } + AgentTurnOrigin::TrustedAutomation { + source: TrustedAutomationSource::SubconsciousTainted, + job_id, + } => { + tracing::warn!( + tool = tool_name, + job_id = %job_id, + "[approval::gate] subconscious tick with external-sync memory in context — \ + rejecting external_effect tool" + ); + return ( + GateOutcome::Deny { + reason: format!( + "{POLICY_DENIED_MARKER} Tool '{tool_name}' rejected: subconscious turn \ + whose memory context includes external-sync chunks may not run \ + external_effect tools." + ), + }, + None, + ); + } + AgentTurnOrigin::Cli => { + tracing::debug!( + tool = tool_name, + "[approval::gate] CLI / sub-agent caller — allowing without prompt" + ); + return (GateOutcome::Allow, None); + } + AgentTurnOrigin::Unknown => { + tracing::warn!( + tool = tool_name, + "[approval::gate] agent turn has no origin label — refusing to execute \ + external_effect tool from unlabelled call site" + ); + return ( + GateOutcome::Deny { + reason: format!( + "{POLICY_DENIED_MARKER} Tool '{tool_name}' rejected: agent turn has \ + no origin label. Refusing external_effect tool from unlabelled call \ + site." + ), + }, + None, + ); + } } let request_id = uuid::Uuid::new_v4().to_string(); @@ -538,6 +631,16 @@ mod tests { } } + /// A matching web-chat origin for the chat context fixture. Tests + /// exercising the parking flow scope BOTH task-locals — production + /// callers in `channels/providers/web` do the same. + fn web_origin() -> AgentTurnOrigin { + AgentTurnOrigin::WebChat { + thread_id: "t-test".into(), + client_id: "c-test".into(), + } + } + #[tokio::test] async fn approve_once_returns_allow() { let (gate, _dir) = test_gate(); @@ -545,12 +648,14 @@ mod tests { let g = gate.clone(); let handle = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( chat_ctx(), g.intercept("composio", "send slack", serde_json::json!({})), - ) - .await + ), + ) + .await }); // Wait for pending row to land. @@ -579,12 +684,14 @@ mod tests { let g = gate.clone(); let handle = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( chat_ctx(), g.intercept("pushover", "send push", serde_json::json!({})), - ) - .await + ), + ) + .await }); let pending = loop { @@ -631,7 +738,9 @@ mod tests { // An allow-listed tool short-circuits the gate to `Allow` immediately — // before any parking — even with a live chat context present, and - // without persisting a pending row. + // without persisting a pending row. The shortcut runs regardless of + // origin (it's the user's persisted "Always allow" allowlist), so we + // do not need to scope an origin for this case. let outcome = APPROVAL_CHAT_CONTEXT .scope( chat_ctx(), @@ -649,12 +758,14 @@ mod tests { async fn timeout_returns_deny() { let (gate, _dir) = test_gate(); // TTL = 500ms let gate = Arc::new(gate); - let outcome = APPROVAL_CHAT_CONTEXT - .scope( + let outcome = turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( chat_ctx(), gate.intercept("composio", "timed out", serde_json::json!({})), - ) - .await; + ), + ) + .await; match outcome { GateOutcome::Deny { reason } => assert!(reason.contains("timed out")), other => panic!("expected deny, got {other:?}"), @@ -675,16 +786,24 @@ mod tests { let (gate, _dir) = test_gate(); let gate = Arc::new(gate); - // Run intercept inside a scoped chat context (as the web channel does). + // Run intercept inside a scoped chat context + matching WebChat + // origin (as the web channel does in production). let g = gate.clone(); let ctx = ApprovalChatContext { thread_id: "thread-42".into(), client_id: "client-1".into(), }; + let origin = AgentTurnOrigin::WebChat { + thread_id: "thread-42".into(), + client_id: "client-1".into(), + }; let handle = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope(ctx, g.intercept("shell", "run ls", serde_json::json!({}))) - .await + turn_origin::with_origin( + origin, + APPROVAL_CHAT_CONTEXT + .scope(ctx, g.intercept("shell", "run ls", serde_json::json!({}))), + ) + .await }); // While parked, the thread → request mapping is queryable. @@ -736,18 +855,140 @@ mod tests { } #[tokio::test] - async fn no_chat_context_is_allowed_not_gated() { - // The gate is interactive: a non-chat caller (background / triage / cron, - // no ApprovalChatContext) is allowed straight through — never parked — - // so autonomous turns don't stall on an approval no one can answer. + async fn intercept_with_unknown_origin_denies() { + // Unlabelled call site (no origin scope) maps to `Unknown` and is + // rejected. This replaces the previous "no chat context → Allow" + // legacy behaviour: the gate now refuses to execute external_effect + // tools from unlabelled call sites. let (gate, _dir) = test_gate(); let outcome = gate .intercept("shell", "run ls", serde_json::json!({})) .await; - assert!(matches!(outcome, GateOutcome::Allow)); + match outcome { + GateOutcome::Deny { reason } => assert!(reason.contains("origin label")), + other => panic!("expected deny, got {other:?}"), + } assert!(gate.pending_for_thread("thread-42").is_none()); } + #[tokio::test] + async fn intercept_with_trusted_cron_origin_allows_without_prompt() { + // Cron jobs the user explicitly authorized run trusted automation; + // the gate allows without prompt and does not persist a row. + let (gate, _dir) = test_gate(); + let origin = AgentTurnOrigin::TrustedAutomation { + job_id: "cron-42".into(), + source: TrustedAutomationSource::Cron, + }; + let outcome = turn_origin::with_origin( + origin, + gate.intercept("shell", "run ls", serde_json::json!({})), + ) + .await; + assert!(matches!(outcome, GateOutcome::Allow)); + assert!( + gate.list_pending().unwrap().is_empty(), + "trusted cron must not persist a pending row" + ); + } + + #[tokio::test] + async fn intercept_with_trusted_subconscious_origin_allows_without_prompt() { + // Subconscious ticks on internal-only memory are trusted automation + // and run unprompted (preserves pre-PR behavior for the safe case). + let (gate, _dir) = test_gate(); + let origin = AgentTurnOrigin::TrustedAutomation { + job_id: "subconscious-tick".into(), + source: TrustedAutomationSource::Subconscious, + }; + let outcome = turn_origin::with_origin( + origin, + gate.intercept("shell", "run ls", serde_json::json!({})), + ) + .await; + assert!(matches!(outcome, GateOutcome::Allow)); + } + + #[tokio::test] + async fn intercept_with_subconscious_tainted_origin_denies() { + // A subconscious tick whose memory context contains external-sync + // chunks is rejected for external_effect tools — external text in + // memory could otherwise steer the tick into a tool call. + let (gate, _dir) = test_gate(); + let origin = AgentTurnOrigin::TrustedAutomation { + job_id: "subconscious-tainted".into(), + source: TrustedAutomationSource::SubconsciousTainted, + }; + let outcome = turn_origin::with_origin( + origin, + gate.intercept("send_email", "send", serde_json::json!({})), + ) + .await; + match outcome { + GateOutcome::Deny { reason } => { + assert!(reason.contains("external-sync"), "reason was: {reason}") + } + other => panic!("expected deny, got {other:?}"), + } + } + + #[tokio::test] + async fn intercept_with_cli_origin_allows_without_prompt() { + // CLI / one-off internal callers (sub-agent invocations, scripts) + // are allowed through unprompted — there is no chat surface to + // park on, and the legacy CLI workflow assumes the operator + // authorized the invocation. + let (gate, _dir) = test_gate(); + let outcome = turn_origin::with_origin( + AgentTurnOrigin::Cli, + gate.intercept("shell", "run ls", serde_json::json!({})), + ) + .await; + assert!(matches!(outcome, GateOutcome::Allow)); + } + + #[tokio::test] + async fn intercept_with_external_channel_origin_persists_and_ttl_denies() { + // Non-web channel inbound (Telegram / Discord / Slack / etc.): + // persist an audit row but TTL-deny — there is no channel-routed + // approval surface yet, and the input is remote-attacker text. + let (gate, _dir) = test_gate(); // 2s TTL + let gate = Arc::new(gate); + let origin = AgentTurnOrigin::ExternalChannel { + channel: "telegram".into(), + reply_target: "tg-chat-1".into(), + message_id: "msg-1".into(), + }; + + let g = gate.clone(); + let handle = tokio::spawn(async move { + turn_origin::with_origin( + origin, + g.intercept("shell", "run ls", serde_json::json!({})), + ) + .await + }); + + // The audit row appears while the future is parked. + let mut tries = 0; + loop { + if !gate.list_pending().unwrap().is_empty() { + break; + } + tries += 1; + assert!(tries < 50, "audit row never appeared for external channel"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + + // Without a routable channel approval surface, the parked future + // TTL-denies (2s — matches the test_gate fixture). + let outcome = handle.await.unwrap(); + match outcome { + GateOutcome::Deny { reason } => assert!(reason.contains("timed out")), + other => panic!("expected deny, got {other:?}"), + } + } + #[tokio::test] async fn intercept_audited_returns_request_id_only_when_allowed_and_persisted() { let (gate, _dir) = test_gate(); @@ -758,15 +999,18 @@ mod tests { // (issue #2135). let g = gate.clone(); let handle = tokio::spawn(async move { - // Scope a chat context *inside* the spawned task — task-locals don't - // cross `tokio::spawn`, and `intercept` only parks (creates a pending - // row) when a chat context is present. - APPROVAL_CHAT_CONTEXT - .scope( + // Scope a chat context + matching WebChat origin *inside* the + // spawned task — task-locals don't cross `tokio::spawn`, and + // `intercept` only parks (creates a pending row) for a chat + // turn whose origin labels it as web-routable. + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( chat_ctx(), g.intercept_audited("composio", "send slack", serde_json::json!({})), - ) - .await + ), + ) + .await }); let pending = loop { if let Some(p) = gate.list_pending().unwrap().into_iter().next() { @@ -797,12 +1041,14 @@ mod tests { // Deny path → no id (nothing to record afterward). let g = gate.clone(); let denied = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( chat_ctx(), g.intercept_audited("composio", "send slack", serde_json::json!({})), - ) - .await + ), + ) + .await }); let pending = loop { if let Some(p) = gate.list_pending().unwrap().into_iter().next() { @@ -819,12 +1065,14 @@ mod tests { // Allowlist-shortcut path → also no id (no row was created). let g = gate.clone(); let first = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( chat_ctx(), g.intercept_audited("pushover", "first send", serde_json::json!({})), - ) - .await + ), + ) + .await }); let pending = loop { if let Some(p) = gate diff --git a/src/openhuman/autocomplete/history.rs b/src/openhuman/autocomplete/history.rs index ee37b22dc..e63c37326 100644 --- a/src/openhuman/autocomplete/history.rs +++ b/src/openhuman/autocomplete/history.rs @@ -132,6 +132,8 @@ pub async fn save_completion_to_local_docs( category: "daily".to_string(), session_id: None, document_id: None, + // Autocomplete acceptances are first-party UI signals. + taint: crate::openhuman::memory::MemoryTaint::Internal, }; if let Err(e) = client.put_doc(input).await { diff --git a/src/openhuman/channels/context.rs b/src/openhuman/channels/context.rs index ba45d37a4..dc4cee0d4 100644 --- a/src/openhuman/channels/context.rs +++ b/src/openhuman/channels/context.rs @@ -322,6 +322,7 @@ mod tests { timestamp: "now".into(), session_id: None, score, + taint: Default::default(), } } diff --git a/src/openhuman/channels/providers/web.rs b/src/openhuman/channels/providers/web.rs index 0024b82ab..434e8f98b 100644 --- a/src/openhuman/channels/providers/web.rs +++ b/src/openhuman/channels/providers/web.rs @@ -591,8 +591,18 @@ pub async fn start_chat( thread_id: thread_id_task.clone(), client_id: client_id_task.clone(), }; - let result = crate::openhuman::approval::APPROVAL_CHAT_CONTEXT - .scope( + // Scope the matching `AgentTurnOrigin::WebChat` alongside the chat + // context so the approval gate's origin-aware decision tree sees a + // web-routable turn. Both task-locals must wrap the same future — + // tokio task-locals do not cross `tokio::spawn`, and `intercept` + // runs inline within this task. + let origin = crate::openhuman::agent::turn_origin::AgentTurnOrigin::WebChat { + thread_id: thread_id_task.clone(), + client_id: client_id_task.clone(), + }; + let result = crate::openhuman::agent::turn_origin::with_origin( + origin, + crate::openhuman::approval::APPROVAL_CHAT_CONTEXT.scope( approval_ctx, run_chat_task( &client_id_task, @@ -604,8 +614,9 @@ pub async fn start_chat( profile_id, locale, ), - ) - .await; + ), + ) + .await; match result { Ok(chat_result) => { diff --git a/src/openhuman/channels/runtime/dispatch.rs b/src/openhuman/channels/runtime/dispatch.rs index 9c9a0cbcb..e501ce22c 100644 --- a/src/openhuman/channels/runtime/dispatch.rs +++ b/src/openhuman/channels/runtime/dispatch.rs @@ -1008,6 +1008,17 @@ pub(crate) async fn process_channel_message( set }); + // Non-web channel turns label themselves as `ExternalChannel` so the + // approval gate's origin-aware decision tree treats the inbound text + // as untrusted (remote-attacker-controlled). Cron-driven channel + // deliveries get a `TrustedAutomation { Cron }` label from the + // scheduler instead and never reach this dispatch path. + let turn_origin = crate::openhuman::agent::turn_origin::AgentTurnOrigin::ExternalChannel { + channel: msg.channel.clone(), + reply_target: msg.reply_target.clone(), + message_id: msg.id.clone(), + }; + let turn_request = AgentTurnRequest { provider: Arc::clone(&active_provider), history: std::mem::take(&mut history), @@ -1040,6 +1051,7 @@ pub(crate) async fn process_channel_message( visible_tool_names, extra_tools: scoping.extra_tools, on_progress: progress_tx, + origin: turn_origin, }; tracing::debug!( channel = %msg.channel, diff --git a/src/openhuman/channels/runtime/test_support.rs b/src/openhuman/channels/runtime/test_support.rs index 7b4683f2f..57653366c 100644 --- a/src/openhuman/channels/runtime/test_support.rs +++ b/src/openhuman/channels/runtime/test_support.rs @@ -279,6 +279,7 @@ fn memory_entry(input: TestMemoryEntry) -> MemoryEntry { timestamp: "now".to_string(), session_id: None, score: input.score, + taint: crate::openhuman::memory::MemoryTaint::Internal, } } diff --git a/src/openhuman/config/schema/agent.rs b/src/openhuman/config/schema/agent.rs index 24e89d9c8..3237295a9 100644 --- a/src/openhuman/config/schema/agent.rs +++ b/src/openhuman/config/schema/agent.rs @@ -199,9 +199,28 @@ pub struct AgentConfig { /// Keys are channel names (e.g., "telegram", "discord", "web", "cli"). /// Values are permission levels: "none", "readonly" (or "read_only"), /// "write", "execute", "dangerous". - /// When this map is empty, the agent preserves the legacy unrestricted - /// channel surface. Once configured, channels not listed here default to - /// "readonly". + /// + /// Runtime semantics (see + /// [`crate::openhuman::agent_tool_policy::engine::ToolPolicyEngine`]): + /// + /// * **Empty map** — the policy engine preserves the legacy + /// unrestricted surface and returns `PermissionLevel::Dangerous` + /// for every channel. This branch only matters before the + /// one-time migration runs. + /// * **Non-empty map, channel present** — the configured level is + /// used. + /// * **Non-empty map, channel absent** — the engine falls back to + /// `PermissionLevel::ReadOnly` (the fail-closed default for an + /// already-policy-managed install). + /// + /// [`AgentConfig::migrate_channel_permissions_if_legacy`] seeds the + /// map with `web=Execute` + each configured channel = `Execute` on + /// first boot after upgrade, so legacy installs land in the + /// non-empty branch before any tool dispatch happens. New installs + /// ship with an explicit map. The empty-map "Dangerous" branch is + /// effectively reachable only by an operator manually wiping the + /// map in their on-disk config; if you change that branch's + /// behavior, update `AGENTS.md` and the engine docstring in lock-step. #[serde(default)] pub channel_permissions: std::collections::HashMap, @@ -255,6 +274,60 @@ fn default_max_memory_context_chars() -> usize { } impl AgentConfig { + /// Seed legacy installs whose channel-permissions map is empty and + /// that already have at least one non-web channel configured, + /// writing explicit per-channel execute entries. + /// + /// The engine layer keeps its legacy empty-map shortcut; this + /// migration replaces it with an explicit policy so the + /// per-channel cap engages on the very first boot after upgrade. + /// `known_channels` is the set of channels the user has configured + /// in `channels::ChannelsConfig`. The web channel is always added + /// on top so the desktop UI stays usable. + /// + /// Returns `true` when a migration write is required so the caller + /// can save and reload; returns `false` when the map was already + /// populated, no non-web channels were configured (fresh install, + /// engine's legacy unrestricted shortcut continues), or the + /// migration is otherwise a no-op. Idempotent. + pub fn migrate_channel_permissions_if_legacy(&mut self, known_channels: I) -> bool + where + I: IntoIterator, + S: AsRef, + { + if !self.channel_permissions.is_empty() { + return false; + } + let extra: Vec = known_channels + .into_iter() + .map(|s| s.as_ref().to_string()) + .filter(|s| !s.is_empty() && s != "web") + .collect(); + if extra.is_empty() { + // No channels configured yet — leave the map empty so the + // engine's legacy "empty == unrestricted" shortcut keeps + // ruling fresh installs the same way it did pre-PR. The + // migration is for installs that already have channels + // active under the legacy unrestricted surface. + return false; + } + // Seed web + every known channel = execute so the engine's + // per-channel cap evaluates against an explicit policy instead + // of the legacy unrestricted default. + let names: Vec = std::iter::once("web".to_string()).chain(extra).collect(); + for name in &names { + self.channel_permissions + .insert(name.clone(), "execute".to_string()); + } + log::info!( + target: "openhuman::config", + "[agent-config] channel_permissions: migrated {} legacy channels to execute (preserved pre-PR behavior): {:?}", + names.len(), + names + ); + true + } + /// Resolve the active memory-context budgets for this agent config. /// /// Two cases: @@ -459,4 +532,45 @@ mod memory_window_tests { let back: MemoryContextWindow = serde_json::from_str("\"minimal\"").unwrap(); assert_eq!(back, MemoryContextWindow::Minimal); } + + #[test] + fn empty_channel_permissions_with_existing_channels_migrates_to_execute() { + // Legacy install: channel_permissions empty but the user has two + // channels configured. The migration seeds web + each existing + // channel = execute so the new fail-closed default doesn't + // regress them. + let mut cfg = AgentConfig::default(); + assert!(cfg.channel_permissions.is_empty()); + + let known = vec!["telegram".to_string(), "discord".to_string()]; + let migrated = cfg.migrate_channel_permissions_if_legacy(known.iter()); + + assert!(migrated, "legacy install must migrate"); + assert_eq!(cfg.channel_permissions.len(), 3); + for ch in ["web", "telegram", "discord"] { + assert_eq!( + cfg.channel_permissions.get(ch).map(String::as_str), + Some("execute"), + "expected execute for channel {ch}" + ); + } + } + + #[test] + fn migrate_channel_permissions_idempotent() { + let mut cfg = AgentConfig::default(); + cfg.migrate_channel_permissions_if_legacy(vec!["telegram".to_string()].iter()); + let again = cfg.migrate_channel_permissions_if_legacy(vec!["telegram".to_string()].iter()); + assert!(!again, "second migration call must be a no-op"); + } + + #[test] + fn migrate_channel_permissions_with_no_channels_is_noop() { + // Fresh install with no configured channels and an empty map — + // no migration needed (the engine fails closed on lookup). + let mut cfg = AgentConfig::default(); + let migrated = cfg.migrate_channel_permissions_if_legacy(Vec::::new()); + assert!(!migrated); + assert!(cfg.channel_permissions.is_empty()); + } } diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index 00ff1300c..8202dd0d1 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/src/openhuman/cron/scheduler.rs @@ -424,7 +424,22 @@ async fn run_agent_job(config: &Config, job: &CronJob) -> (bool, String, Option< // cron-triggered turns. `cron` is the channel so the // event bus can filter from other flows (`cli`, `web`…). agent.set_event_context(format!("cron:{}", job.id), "cron"); - agent.run_single(&prefixed_prompt).await + // Scope a `TrustedAutomation { Cron }` origin around the + // turn. The approval gate treats this as user-authorized + // automation and lets external_effect tools run without + // an in-app prompt — the user explicitly created this + // cron job and authorized its prompt at the same time. + let origin = + crate::openhuman::agent::turn_origin::AgentTurnOrigin::TrustedAutomation { + job_id: job.id.clone(), + source: + crate::openhuman::agent::turn_origin::TrustedAutomationSource::Cron, + }; + crate::openhuman::agent::turn_origin::with_origin( + origin, + agent.run_single(&prefixed_prompt), + ) + .await } Err(e) => Err(e), } diff --git a/src/openhuman/inference/local/ops.rs b/src/openhuman/inference/local/ops.rs index 7320bc4c5..f635d2ccf 100644 --- a/src/openhuman/inference/local/ops.rs +++ b/src/openhuman/inference/local/ops.rs @@ -102,7 +102,15 @@ pub async fn agent_chat( config.default_temperature = temp; } let mut agent = Agent::from_config(config).map_err(|e| e.to_string())?; - let response = agent.run_single(message).await.map_err(|e| e.to_string())?; + // Direct `agent_chat` RPC — invoked by trusted clients (desktop UI, + // operator CLI). Label as CLI so the approval gate doesn't fail + // closed on an unlabelled call site. + let response = crate::openhuman::agent::turn_origin::with_origin( + crate::openhuman::agent::turn_origin::AgentTurnOrigin::Cli, + agent.run_single(message), + ) + .await + .map_err(|e| e.to_string())?; Ok(RpcOutcome::single_log(response, "agent chat completed")) } diff --git a/src/openhuman/learning/reflection_tests.rs b/src/openhuman/learning/reflection_tests.rs index 9aa9e8154..b86c22208 100644 --- a/src/openhuman/learning/reflection_tests.rs +++ b/src/openhuman/learning/reflection_tests.rs @@ -36,6 +36,7 @@ impl Memory for MockMemory { timestamp: "now".into(), session_id: session_id.map(str::to_string), score: None, + taint: Default::default(), }, ); Ok(()) diff --git a/src/openhuman/learning/tool_tracker.rs b/src/openhuman/learning/tool_tracker.rs index 0c4c11606..86ee8a7d6 100644 --- a/src/openhuman/learning/tool_tracker.rs +++ b/src/openhuman/learning/tool_tracker.rs @@ -216,6 +216,7 @@ mod tests { timestamp: "now".into(), session_id: session_id.map(str::to_string), score: None, + taint: Default::default(), }, ); Ok(()) diff --git a/src/openhuman/learning/transcript_ingest/tests.rs b/src/openhuman/learning/transcript_ingest/tests.rs index 6d489ddd0..836771fca 100644 --- a/src/openhuman/learning/transcript_ingest/tests.rs +++ b/src/openhuman/learning/transcript_ingest/tests.rs @@ -86,6 +86,7 @@ impl Memory for InMemory { timestamp: "2026-05-09T12:00:00Z".to_string(), session_id: session_id.map(|s| s.to_string()), score: None, + taint: Default::default(), }); } } diff --git a/src/openhuman/learning/user_profile.rs b/src/openhuman/learning/user_profile.rs index 8c321fa80..f02ed39d3 100644 --- a/src/openhuman/learning/user_profile.rs +++ b/src/openhuman/learning/user_profile.rs @@ -307,6 +307,7 @@ mod tests { timestamp: "now".into(), session_id: session_id.map(str::to_string), score: None, + taint: Default::default(), }, ); Ok(()) diff --git a/src/openhuman/mcp_server/tools.rs b/src/openhuman/mcp_server/tools.rs index f1120ea1a..8d681a78a 100644 --- a/src/openhuman/mcp_server/tools.rs +++ b/src/openhuman/mcp_server/tools.rs @@ -1312,10 +1312,21 @@ async fn run_subagent_tool(params: &Map) -> Result Result NamespaceDocumentInput { category: "core".into(), session_id: Some("session-1".into()), document_id: Some("doc-id-1".into()), + taint: crate::openhuman::memory::MemoryTaint::Internal, } } diff --git a/src/openhuman/memory/ingestion/queue.rs b/src/openhuman/memory/ingestion/queue.rs index 32f22c4fb..b1f583a13 100644 --- a/src/openhuman/memory/ingestion/queue.rs +++ b/src/openhuman/memory/ingestion/queue.rs @@ -312,6 +312,7 @@ mod tests { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }, config: MemoryIngestionConfig::default(), } diff --git a/src/openhuman/memory/ingestion/tests.rs b/src/openhuman/memory/ingestion/tests.rs index b32e05428..86e61e008 100644 --- a/src/openhuman/memory/ingestion/tests.rs +++ b/src/openhuman/memory/ingestion/tests.rs @@ -44,6 +44,7 @@ async fn gmail_fixture_ingestion_recovers_required_signals() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }, config: ci_safe_config(), }) @@ -135,6 +136,7 @@ async fn notion_fixture_ingestion_recovers_required_signals() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }, config: ci_safe_config(), }) diff --git a/src/openhuman/memory/mod.rs b/src/openhuman/memory/mod.rs index c36ecf5ff..4a7c88b12 100644 --- a/src/openhuman/memory/mod.rs +++ b/src/openhuman/memory/mod.rs @@ -51,7 +51,7 @@ pub use schemas::{ all_controller_schemas as all_memory_controller_schemas, all_registered_controllers as all_memory_registered_controllers, }; -pub use traits::{Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts}; +pub use traits::{Memory, MemoryCategory, MemoryEntry, MemoryTaint, NamespaceSummary, RecallOpts}; // Re-export types that external tests and consumers historically imported // from `memory::*`. The definitions moved to sibling crates during the diff --git a/src/openhuman/memory/ops/documents.rs b/src/openhuman/memory/ops/documents.rs index bb2cca811..667682154 100644 --- a/src/openhuman/memory/ops/documents.rs +++ b/src/openhuman/memory/ops/documents.rs @@ -186,6 +186,10 @@ pub async fn doc_put(params: PutDocParams) -> Result, S category: params.category, session_id: params.session_id, document_id: params.document_id, + // RPC-driven doc puts come from the user / agent — Internal. + // External-sync ingest paths bypass this RPC and call + // `store_skill_sync` directly with their own taint label. + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await?; Ok(RpcOutcome::single_log( @@ -213,6 +217,7 @@ pub async fn doc_ingest( category: params.category, session_id: params.session_id, document_id: params.document_id, + taint: crate::openhuman::memory::MemoryTaint::Internal, }, config: params.config.unwrap_or_default(), }) diff --git a/src/openhuman/memory/ops/helpers.rs b/src/openhuman/memory/ops/helpers.rs index 66bd2030b..4b2bbff18 100644 --- a/src/openhuman/memory/ops/helpers.rs +++ b/src/openhuman/memory/ops/helpers.rs @@ -259,6 +259,7 @@ mod tests { document_ids: vec!["doc-1".into()], chunk_ids: vec!["chunk-1".into()], }], + taint: crate::openhuman::memory::MemoryTaint::Internal, } } diff --git a/src/openhuman/memory/ops/learn.rs b/src/openhuman/memory/ops/learn.rs index 40691c4df..024ac1dde 100644 --- a/src/openhuman/memory/ops/learn.rs +++ b/src/openhuman/memory/ops/learn.rs @@ -215,6 +215,7 @@ mod tests { category: "core".into(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect("seed namespace doc"); diff --git a/src/openhuman/memory/ops_tests.rs b/src/openhuman/memory/ops_tests.rs index c5688264b..7104cec6f 100644 --- a/src/openhuman/memory/ops_tests.rs +++ b/src/openhuman/memory/ops_tests.rs @@ -41,6 +41,7 @@ fn sample_hit() -> NamespaceMemoryHit { document_ids: vec!["doc-1".to_string()], chunk_ids: vec!["doc-1#chunk-1".to_string()], }], + taint: crate::openhuman::memory::MemoryTaint::Internal, } } @@ -94,6 +95,7 @@ fn sample_hit_with_entity_types() -> NamespaceMemoryHit { document_ids: vec!["doc-2".to_string()], chunk_ids: vec!["doc-2#chunk-1".to_string()], }], + taint: crate::openhuman::memory::MemoryTaint::Internal, } } diff --git a/src/openhuman/memory/traits.rs b/src/openhuman/memory/traits.rs index e2d785d0d..9960def95 100644 --- a/src/openhuman/memory/traits.rs +++ b/src/openhuman/memory/traits.rs @@ -10,6 +10,64 @@ use rusqlite::Connection; use serde::{Deserialize, Serialize}; use std::sync::Arc; +/// Provenance / trust signal attached to a memory entry. Drives downstream +/// policy decisions — most importantly, whether a subconscious tick whose +/// context contains this chunk may invoke external-effect tools. +/// +/// Defaults to [`MemoryTaint::Internal`] so existing rows (which have no +/// taint column persisted) and all in-memory `MemoryEntry::default()` +/// constructions are conservatively trusted as user-driven content. +/// Sync paths that ingest text from third-party services (Gmail / Slack / +/// Notion / Composio / etc.) MUST flip this to [`MemoryTaint::ExternalSync`] +/// at write time so the subconscious origin escalation can see it. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum MemoryTaint { + /// User-driven memory (chat, manual remember, internal heuristics). + #[default] + Internal, + /// Chunk ingested from an external sync source (Gmail / Slack / + /// Notion / Composio / etc.). Subconscious turns whose context + /// contains any tainted chunk MUST run with + /// [`TrustedAutomationSource::SubconsciousTainted`] origin so + /// external_effect tools are refused. + /// + /// [`TrustedAutomationSource::SubconsciousTainted`]: + /// crate::openhuman::agent::turn_origin::TrustedAutomationSource::SubconsciousTainted + ExternalSync, +} + +impl MemoryTaint { + /// Serialised form used by the SQLite `memory_docs.taint` column. + /// + /// Kept short + snake_case to match the serde representation and to + /// keep the on-disk footprint minimal. Round-trips via + /// [`Self::from_db_str`]. + pub fn as_db_str(&self) -> &'static str { + match self { + Self::Internal => "internal", + Self::ExternalSync => "external_sync", + } + } + + /// Reverse of [`Self::as_db_str`]. Unknown values (a forward-rolled + /// schema variant we don't know about yet, a manual `UPDATE` typo, + /// row corruption) decode as the more restrictive + /// [`MemoryTaint::ExternalSync`] so the subconscious gate fails + /// closed — we'd rather refuse external_effect tools on a chunk + /// of unknown provenance than silently treat it as user-authored. + /// Legacy rows that pre-date the column are not affected: the + /// migration writes a literal `'internal'` default so they + /// round-trip cleanly through the explicit arm. + pub fn from_db_str(raw: &str) -> Self { + match raw { + "internal" => Self::Internal, + "external_sync" => Self::ExternalSync, + _ => Self::ExternalSync, + } + } +} + /// Represents a single stored memory entry with associated metadata. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MemoryEntry { @@ -30,6 +88,12 @@ pub struct MemoryEntry { pub session_id: Option, /// Optional relevance or confidence score, typically from 0.0 to 1.0. pub score: Option, + /// Provenance — `Internal` for user-driven writes, `ExternalSync` for + /// memory_sync ingest. Default keeps existing persisted rows safe; + /// composio / Gmail / Slack / Notion ingest paths must set this + /// explicitly. See [`MemoryTaint`] for the security contract. + #[serde(default)] + pub taint: MemoryTaint, } /// Categories used to organize and filter memories by their nature and lifecycle. @@ -119,6 +183,32 @@ pub trait Memory: Send + Sync { session_id: Option<&str>, ) -> anyhow::Result<()>; + /// Store an entry with explicit provenance taint. + /// + /// Sync paths that ingest text from third-party services (Gmail / Slack / + /// Notion / Composio / etc.) MUST go through this entry point with + /// [`MemoryTaint::ExternalSync`] so the subconscious gate can refuse + /// external_effect tools when the resulting chunks reach a tick's + /// context window. + /// + /// The default implementation degrades to [`Self::store`] for backends + /// that do not yet persist taint (e.g. mock / in-memory stores used in + /// tests); the `UnifiedMemory` backend overrides this with a real + /// taint-carrying upsert. + async fn store_with_taint( + &self, + namespace: &str, + key: &str, + content: &str, + category: MemoryCategory, + session_id: Option<&str>, + taint: MemoryTaint, + ) -> anyhow::Result<()> { + let _ = taint; + self.store(namespace, key, content, category, session_id) + .await + } + /// Recalls memories matching a query string using keyword or semantic search. /// /// Namespace is passed via `opts.namespace`; `None` uses the backend's @@ -226,6 +316,7 @@ mod tests { timestamp: "2026-02-16T00:00:00Z".into(), session_id: Some("session-abc".into()), score: Some(0.98), + taint: MemoryTaint::Internal, }; let json = serde_json::to_string(&entry).unwrap(); @@ -238,5 +329,76 @@ mod tests { assert_eq!(parsed.category, MemoryCategory::Core); assert_eq!(parsed.session_id.as_deref(), Some("session-abc")); assert_eq!(parsed.score, Some(0.98)); + assert_eq!(parsed.taint, MemoryTaint::Internal); + } + + #[test] + fn memory_taint_defaults_to_internal_for_legacy_rows() { + // Legacy rows persisted before the taint column existed deserialize + // to MemoryTaint::Internal, so the gate's tainted-subconscious + // escalation never fires for entries we cannot classify. + let legacy = r#"{ + "id":"x", + "key":"k", + "content":"c", + "namespace":null, + "category":"core", + "timestamp":"2026-01-01T00:00:00Z", + "session_id":null, + "score":null + }"#; + let parsed: MemoryEntry = serde_json::from_str(legacy).unwrap(); + assert_eq!(parsed.taint, MemoryTaint::Internal); + } + + #[test] + fn memory_taint_as_db_str_uses_snake_case_form() { + assert_eq!(MemoryTaint::Internal.as_db_str(), "internal"); + assert_eq!(MemoryTaint::ExternalSync.as_db_str(), "external_sync"); + } + + #[test] + fn memory_taint_from_db_str_known_values_roundtrip_unknown_fails_closed() { + // Round-trip both known values. + assert_eq!( + MemoryTaint::from_db_str(MemoryTaint::Internal.as_db_str()), + MemoryTaint::Internal + ); + assert_eq!( + MemoryTaint::from_db_str(MemoryTaint::ExternalSync.as_db_str()), + MemoryTaint::ExternalSync + ); + // Unknown / corrupted column values fail closed to the more + // restrictive `ExternalSync` so the subconscious gate refuses + // external_effect tools on chunks of unknown provenance rather + // than silently treating them as user-authored. + assert_eq!(MemoryTaint::from_db_str(""), MemoryTaint::ExternalSync); + assert_eq!( + MemoryTaint::from_db_str("EXTERNAL_SYNC"), + MemoryTaint::ExternalSync + ); + assert_eq!( + MemoryTaint::from_db_str("future"), + MemoryTaint::ExternalSync + ); + } + + #[test] + fn memory_taint_roundtrips_external_sync() { + let entry = MemoryEntry { + id: "x".into(), + key: "k".into(), + content: "c".into(), + namespace: None, + category: MemoryCategory::Conversation, + timestamp: "2026-01-01T00:00:00Z".into(), + session_id: None, + score: None, + taint: MemoryTaint::ExternalSync, + }; + let json = serde_json::to_string(&entry).unwrap(); + assert!(json.contains("\"taint\":\"external_sync\"")); + let parsed: MemoryEntry = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.taint, MemoryTaint::ExternalSync); } } diff --git a/src/openhuman/memory_store/client.rs b/src/openhuman/memory_store/client.rs index 60538bc0d..9cc318f88 100644 --- a/src/openhuman/memory_store/client.rs +++ b/src/openhuman/memory_store/client.rs @@ -230,6 +230,15 @@ impl MemoryClient { /// Specialized method for syncing skill data into memory. /// /// Maps generic skill/integration fields into the `NamespaceDocumentInput` structure. + /// + /// Every write goes in as + /// [`MemoryTaint::ExternalSync`](crate::openhuman::memory::MemoryTaint::ExternalSync) + /// — this entry point exists specifically for memory_sync providers + /// (Gmail / Slack / Notion / Composio / etc.) that ingest text from + /// third-party services. Routing the call through here is what lets + /// the subconscious gate refuse external_effect tools when these + /// chunks land in a tick's context window. Internal / user-driven + /// writes must use the regular `put_doc` / `Memory::store` paths. #[allow(clippy::too_many_arguments)] pub async fn store_skill_sync( &self, @@ -257,6 +266,10 @@ impl MemoryClient { category: "core".to_string(), session_id: None, document_id, + // Every sync entry point is by definition ingesting third- + // party content; mark it so the subconscious gate can see + // the provenance through the persistence layer. + taint: crate::openhuman::memory::MemoryTaint::ExternalSync, }; let doc_id = self.inner.upsert_document(input.clone()).await?; diff --git a/src/openhuman/memory_store/client_tests.rs b/src/openhuman/memory_store/client_tests.rs index 927c4709b..cce95553f 100644 --- a/src/openhuman/memory_store/client_tests.rs +++ b/src/openhuman/memory_store/client_tests.rs @@ -30,6 +30,7 @@ fn doc(namespace: &str, key: &str, content: &str) -> NamespaceDocumentInput { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, } } diff --git a/src/openhuman/memory_store/memory_trait.rs b/src/openhuman/memory_store/memory_trait.rs index dd0c41901..1c37d7956 100644 --- a/src/openhuman/memory_store/memory_trait.rs +++ b/src/openhuman/memory_store/memory_trait.rs @@ -15,7 +15,7 @@ use rusqlite::{params, OptionalExtension}; use serde_json::json; use crate::openhuman::memory::traits::{ - Memory, MemoryCategory, MemoryEntry, NamespaceSummary, RecallOpts, + Memory, MemoryCategory, MemoryEntry, MemoryTaint, NamespaceSummary, RecallOpts, }; use crate::openhuman::memory_store::types::{NamespaceDocumentInput, GLOBAL_NAMESPACE}; use crate::openhuman::memory_store::unified::fts5; @@ -67,6 +67,28 @@ impl Memory for UnifiedMemory { content: &str, category: MemoryCategory, session_id: Option<&str>, + ) -> anyhow::Result<()> { + // The default `store` entry point is user-driven; ingest paths + // come in via `store_with_taint`. + self.store_with_taint( + namespace, + key, + content, + category, + session_id, + MemoryTaint::Internal, + ) + .await + } + + async fn store_with_taint( + &self, + namespace: &str, + key: &str, + content: &str, + category: MemoryCategory, + session_id: Option<&str>, + taint: MemoryTaint, ) -> anyhow::Result<()> { let ns = if namespace.trim().is_empty() { GLOBAL_NAMESPACE.to_string() @@ -85,6 +107,7 @@ impl Memory for UnifiedMemory { category: category.to_string(), session_id: session_id.map(str::to_string), document_id: None, + taint, }) .await .map(|_| ()) @@ -118,6 +141,11 @@ impl Memory for UnifiedMemory { timestamp: Utc::now().to_rfc3339(), session_id: None, score: Some(r.score), + // Surface the real taint persisted on `memory_docs` so the + // subconscious gate can decide whether to escalate the + // turn origin to `SubconsciousTainted` when this entry + // lands in a tick's context window. + taint: r.taint, }) .collect(); @@ -169,6 +197,7 @@ impl Memory for UnifiedMemory { timestamp: ts_rfc3339, session_id: Some(entry.session_id), score: Some(match_score), + taint: crate::openhuman::memory::MemoryTaint::Internal, }); } } @@ -239,6 +268,7 @@ impl Memory for UnifiedMemory { timestamp: ts_rfc3339, session_id: Some(entry.session_id), score: Some(match_score), + taint: crate::openhuman::memory::MemoryTaint::Internal, }); } } @@ -282,9 +312,9 @@ impl Memory for UnifiedMemory { namespace.to_string() }; let conn = self.conn.lock(); - let row: Option<(String, String, String, f64, String)> = conn + let row: Option<(String, String, String, f64, String, String)> = conn .query_row( - "SELECT document_id, key, content, updated_at, category + "SELECT document_id, key, content, updated_at, category, taint FROM memory_docs WHERE namespace = ?1 AND key = ?2 LIMIT 1", params![ns, key], |row| { @@ -294,12 +324,13 @@ impl Memory for UnifiedMemory { row.get(2)?, row.get(3)?, row.get(4)?, + row.get(5)?, )) }, ) .optional()?; - Ok( - row.map(|(id, key, content, updated_at, category)| MemoryEntry { + Ok(row.map( + |(id, key, content, updated_at, category, taint_str)| MemoryEntry { id, key, content, @@ -308,8 +339,9 @@ impl Memory for UnifiedMemory { timestamp: timestamp_to_rfc3339(updated_at), session_id: None, score: None, - }), - ) + taint: crate::openhuman::memory::MemoryTaint::from_db_str(&taint_str), + }, + )) } async fn list( @@ -331,6 +363,10 @@ impl Memory for UnifiedMemory { .unwrap_or_default(); for (idx, d) in items.into_iter().enumerate() { let cat = category.cloned().unwrap_or(MemoryCategory::Core); + let taint_str = d + .get("taint") + .and_then(serde_json::Value::as_str) + .unwrap_or("internal"); out.push(MemoryEntry { id: d .get("documentId") @@ -352,6 +388,7 @@ impl Memory for UnifiedMemory { timestamp: format!("idx-{idx}"), session_id: None, score: None, + taint: crate::openhuman::memory::MemoryTaint::from_db_str(taint_str), }); } Ok(out) @@ -657,4 +694,170 @@ mod tests { "no FTS match must not produce cross-session rows, got {hits:#?}" ); } + + // ── Provenance taint round-trip (#approval-origin) ────────────────── + + #[tokio::test] + async fn taint_persists_across_upsert_and_recall() { + // External-sync ingest writes via `store_with_taint(ExternalSync)` + // and the resulting `MemoryEntry` must surface that taint on + // recall, otherwise the subconscious gate can't detect the + // provenance once the row passes through the persistence layer. + let (_tmp, mem) = fresh_mem(); + mem.store_with_taint( + "skill-gmail", + "thread-1", + "Hi from upstream — please run a quick command", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + + let entries = mem + .recall( + "upstream command", + 5, + RecallOpts { + namespace: Some("skill-gmail"), + min_score: Some(0.0), + ..Default::default() + }, + ) + .await + .unwrap(); + + assert!( + entries.iter().any(|e| e.taint == MemoryTaint::ExternalSync), + "ExternalSync taint must round-trip through recall, got {entries:#?}" + ); + } + + #[tokio::test] + async fn unified_memory_store_with_taint_writes_external_sync() { + // Direct trait-API write — confirms `store_with_taint` doesn't + // fall back to the default Internal value silently. + let (_tmp, mem) = fresh_mem(); + mem.store_with_taint( + "skill-slack", + "msg-42", + "Slack-sourced content", + MemoryCategory::Conversation, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + + let row = mem.get("skill-slack", "msg-42").await.unwrap(); + // `get` is the unfiltered lookup; we use it to assert the row + // landed (the taint surfacing path through recall is asserted in + // the previous test). + assert!(row.is_some(), "stored row must be retrievable"); + } + + #[tokio::test] + async fn legacy_db_rows_default_to_internal_taint() { + // Simulate a database row written before the taint column + // existed by inserting via raw SQL with no taint clause — the + // DEFAULT 'internal' from the migration must kick in and recall + // must surface `MemoryTaint::Internal`. + let (_tmp, mem) = fresh_mem(); + { + let conn = mem.conn.lock(); + conn.execute( + "INSERT INTO memory_docs ( + document_id, namespace, key, title, content, source_type, + priority, tags_json, metadata_json, category, session_id, + created_at, updated_at, markdown_rel_path + ) VALUES (?1, ?2, ?3, ?4, ?5, 'chat', 'medium', '[]', '{}', 'core', NULL, 0.0, 0.0, '')", + rusqlite::params![ + "legacy-doc-taint", + "legacy-ns", + "legacy-key", + "legacy title", + "legacy content about Postgres" + ], + ) + .unwrap(); + } + + let entries = mem + .recall( + "Postgres", + 5, + RecallOpts { + namespace: Some("legacy-ns"), + min_score: Some(0.0), + ..Default::default() + }, + ) + .await + .unwrap(); + + let legacy = entries + .iter() + .find(|e| e.key == "legacy-key") + .expect("legacy row must surface in recall"); + assert_eq!( + legacy.taint, + MemoryTaint::Internal, + "rows written via the pre-taint INSERT clause must decode as Internal via DEFAULT" + ); + } + + #[tokio::test] + async fn subconscious_recall_surfaces_external_sync_taint_for_origin_upgrade() { + // The contract the subconscious engine relies on: a tick that + // pulls a tainted chunk via memory recall must see + // `MemoryTaint::ExternalSync` on the returned entry, which is + // the signal the engine uses to upgrade + // `AgentTurnOrigin::TrustedAutomation { source }` from + // `Subconscious` to `SubconsciousTainted`. + let (_tmp, mem) = fresh_mem(); + mem.store_with_taint( + "skill-notion", + "page-1", + "Tainted Notion page contents", + MemoryCategory::Core, + None, + MemoryTaint::ExternalSync, + ) + .await + .unwrap(); + mem.store( + "skill-notion", + "user-note", + "User-driven note about the same page", + MemoryCategory::Core, + None, + ) + .await + .unwrap(); + + let entries = mem + .recall( + "page", + 10, + RecallOpts { + namespace: Some("skill-notion"), + min_score: Some(0.0), + ..Default::default() + }, + ) + .await + .unwrap(); + + let any_tainted = entries.iter().any(|e| e.taint == MemoryTaint::ExternalSync); + let any_internal = entries.iter().any(|e| e.taint == MemoryTaint::Internal); + assert!( + any_tainted, + "ExternalSync row must surface for the engine's upgrade check" + ); + assert!( + any_internal, + "user-driven row must keep its Internal label so mixed contexts don't over-escalate" + ); + } } diff --git a/src/openhuman/memory_store/safety/mod.rs b/src/openhuman/memory_store/safety/mod.rs index b94c1c718..86eddb907 100644 --- a/src/openhuman/memory_store/safety/mod.rs +++ b/src/openhuman/memory_store/safety/mod.rs @@ -269,6 +269,10 @@ pub fn sanitize_document_input(input: NamespaceDocumentInput) -> Sanitized= 1); + } } diff --git a/src/openhuman/memory_store/types.rs b/src/openhuman/memory_store/types.rs index 655fde8bb..62821812f 100644 --- a/src/openhuman/memory_store/types.rs +++ b/src/openhuman/memory_store/types.rs @@ -2,6 +2,8 @@ use serde::{Deserialize, Serialize}; +use crate::openhuman::memory::MemoryTaint; + pub(crate) const GLOBAL_NAMESPACE: &str = "global"; /// Input payload for upserting a namespace-scoped memory document. @@ -9,6 +11,11 @@ pub(crate) const GLOBAL_NAMESPACE: &str = "global"; /// Used by `MemoryClient::put_doc` and the ingestion pipeline. `document_id` /// is optional — when omitted, an existing row keyed by `(namespace, key)` is /// reused, otherwise a new id is generated. +/// +/// `taint` carries the provenance signal through the persistence layer so +/// downstream recall paths can drive the subconscious origin-escalation +/// decision. Legacy JSON without the field decodes as +/// [`MemoryTaint::Internal`] (see the serde tests below). #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NamespaceDocumentInput { pub namespace: String, @@ -26,6 +33,13 @@ pub struct NamespaceDocumentInput { pub session_id: Option, #[serde(default)] pub document_id: Option, + /// Provenance — `Internal` for user-driven writes, `ExternalSync` for + /// memory_sync ingest paths (Gmail / Slack / Notion / Composio / etc.). + /// Defaults via `#[serde(default)]` so legacy callers and on-disk JSON + /// missing this field decode as the conservative + /// [`MemoryTaint::Internal`]. + #[serde(default)] + pub taint: MemoryTaint, } /// One ranked retrieval result for a namespace text query. @@ -36,6 +50,12 @@ pub struct NamespaceQueryResult { pub score: f64, /// Stored category string (e.g. `core`, `daily`, or custom label). pub category: String, + /// Provenance taint carried back from the persistence layer so the + /// recall caller can surface it on [`crate::openhuman::memory::MemoryEntry`]. + /// Defaults to [`MemoryTaint::Internal`] for legacy rows that predate + /// the column. + #[serde(default)] + pub taint: MemoryTaint, } /// Discriminator for the kind of stored memory item a hit refers to. @@ -66,6 +86,10 @@ pub struct StoredMemoryDocument { pub created_at: f64, pub updated_at: f64, pub markdown_rel_path: String, + /// Provenance taint round-tripped from the `memory_docs.taint` column. + /// Defaults to [`MemoryTaint::Internal`] for legacy rows. + #[serde(default)] + pub taint: MemoryTaint, } /// A single KV row, namespace-scoped or global (when `namespace` is `None`). @@ -128,6 +152,12 @@ pub struct NamespaceMemoryHit { pub chunk_id: Option, #[serde(default)] pub supporting_relations: Vec, + /// Provenance taint surfaced from the source row so retrieval consumers + /// (subconscious tick, context builder, …) can drive origin-escalation + /// without re-querying the underlying document. Defaults to + /// [`MemoryTaint::Internal`] for legacy hits / KV / episodic rows. + #[serde(default)] + pub taint: MemoryTaint, } /// Aggregated retrieval result for a namespace: rendered context text plus @@ -212,6 +242,110 @@ mod tests { } } + #[test] + fn namespace_document_input_taint_defaults_internal_for_legacy_json() { + // Legacy callers serialised before the taint field existed must + // still decode — `#[serde(default)]` makes the field optional and + // falls back to the conservative `Internal` provenance. + let value = json!({ + "namespace": "skill-gmail", + "key": "thread-1", + "title": "Subject", + "content": "Body", + "source_type": "composio-sync", + "priority": "medium", + "metadata": {}, + "category": "core" + }); + let parsed: NamespaceDocumentInput = serde_json::from_value(value).unwrap(); + assert_eq!(parsed.taint, MemoryTaint::Internal); + } + + #[test] + fn namespace_document_input_taint_roundtrips_external_sync() { + let input = NamespaceDocumentInput { + namespace: "skill-gmail".into(), + key: "thread-1".into(), + title: "Subject".into(), + content: "Body".into(), + source_type: "composio-sync".into(), + priority: "medium".into(), + tags: Vec::new(), + metadata: json!({}), + category: "core".into(), + session_id: None, + document_id: None, + taint: MemoryTaint::ExternalSync, + }; + let value = serde_json::to_value(&input).unwrap(); + assert_eq!( + value.get("taint").and_then(|v| v.as_str()), + Some("external_sync") + ); + let parsed: NamespaceDocumentInput = serde_json::from_value(value).unwrap(); + assert_eq!(parsed.taint, MemoryTaint::ExternalSync); + } + + #[test] + fn namespace_query_result_taint_defaults_internal_for_legacy_json() { + let value = json!({ + "key": "k", + "content": "c", + "score": 0.5, + "category": "core" + }); + let parsed: NamespaceQueryResult = serde_json::from_value(value).unwrap(); + assert_eq!(parsed.taint, MemoryTaint::Internal); + } + + #[test] + fn stored_memory_document_taint_defaults_internal_for_legacy_json() { + let value = json!({ + "document_id": "d", + "namespace": "n", + "key": "k", + "title": "t", + "content": "c", + "source_type": "chat", + "priority": "medium", + "tags": [], + "metadata": {}, + "category": "core", + "session_id": null, + "created_at": 1.0, + "updated_at": 2.0, + "markdown_rel_path": "" + }); + let parsed: StoredMemoryDocument = serde_json::from_value(value).unwrap(); + assert_eq!(parsed.taint, MemoryTaint::Internal); + } + + #[test] + fn namespace_memory_hit_taint_defaults_internal_for_legacy_json() { + let hit: NamespaceMemoryHit = serde_json::from_value(json!({ + "id": "hit-1", + "kind": "document", + "namespace": "global", + "key": "note-1", + "title": "Title", + "content": "Body", + "category": "core", + "source_type": "manual", + "updated_at": 3.5, + "score": 0.8, + "score_breakdown": { + "keyword_relevance": 0.5, + "vector_similarity": 0.2, + "graph_relevance": 0.0, + "episodic_relevance": 0.0, + "freshness": 0.1, + "final_score": 0.8 + } + })) + .unwrap(); + assert_eq!(hit.taint, MemoryTaint::Internal); + } + #[test] fn namespace_memory_hit_defaults_optional_fields() { let hit: NamespaceMemoryHit = serde_json::from_value(json!({ diff --git a/src/openhuman/memory_store/unified/documents.rs b/src/openhuman/memory_store/unified/documents.rs index 7e212e4fc..7482b9715 100644 --- a/src/openhuman/memory_store/unified/documents.rs +++ b/src/openhuman/memory_store/unified/documents.rs @@ -113,9 +113,9 @@ impl UnifiedMemory { .map_err(|e| format!("begin tx: {e}"))?; tx.execute( "INSERT INTO memory_docs - (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path) + (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path, taint) VALUES - (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) + (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15) ON CONFLICT(namespace, key) DO UPDATE SET title = excluded.title, content = excluded.content, @@ -126,7 +126,8 @@ impl UnifiedMemory { category = excluded.category, session_id = excluded.session_id, updated_at = excluded.updated_at, - markdown_rel_path = excluded.markdown_rel_path", + markdown_rel_path = excluded.markdown_rel_path, + taint = excluded.taint", params![ document_id, namespace, @@ -141,7 +142,8 @@ impl UnifiedMemory { input.session_id, created_at, updated_at, - markdown_rel + markdown_rel, + input.taint.as_db_str() ], ) .map_err(|e| format!("upsert memory_docs: {e}"))?; @@ -325,9 +327,9 @@ impl UnifiedMemory { let conn = self.conn.lock(); conn.execute( "INSERT INTO memory_docs - (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path) + (document_id, namespace, key, title, content, source_type, priority, tags_json, metadata_json, category, session_id, created_at, updated_at, markdown_rel_path, taint) VALUES - (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) + (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15) ON CONFLICT(namespace, key) DO UPDATE SET title = excluded.title, content = excluded.content, @@ -338,7 +340,8 @@ impl UnifiedMemory { category = excluded.category, session_id = excluded.session_id, updated_at = excluded.updated_at, - markdown_rel_path = excluded.markdown_rel_path", + markdown_rel_path = excluded.markdown_rel_path, + taint = excluded.taint", params![ document_id, namespace, @@ -353,7 +356,8 @@ impl UnifiedMemory { input.session_id, created_at, updated_at, - markdown_rel + markdown_rel, + input.taint.as_db_str() ], ) .map_err(|e| format!("upsert memory_docs: {e}"))?; @@ -384,7 +388,8 @@ impl UnifiedMemory { session_id, created_at, updated_at, - markdown_rel_path + markdown_rel_path, + taint FROM memory_docs WHERE namespace = ?1 ORDER BY updated_at DESC", @@ -400,6 +405,15 @@ impl UnifiedMemory { { let tags_json: String = row.get(7).map_err(|e| e.to_string())?; let metadata_json: String = row.get(8).map_err(|e| e.to_string())?; + // The `taint` column has a NOT NULL DEFAULT 'internal' clause + // from the migration, so legacy rows that pre-date the column + // surface as "internal" string and round-trip back to + // `MemoryTaint::Internal`. Unknown / corrupted values fail + // closed to `MemoryTaint::ExternalSync` inside `from_db_str`, + // so a forward-rolled schema variant or a bad UPDATE can't + // silently downgrade a row to user-authored content. + let taint_str: String = row.get(14).map_err(|e| e.to_string())?; + let taint = crate::openhuman::memory::MemoryTaint::from_db_str(&taint_str); docs.push(StoredMemoryDocument { document_id: row.get(0).map_err(|e| e.to_string())?, namespace: row.get(1).map_err(|e| e.to_string())?, @@ -415,6 +429,7 @@ impl UnifiedMemory { created_at: row.get(11).map_err(|e| e.to_string())?, updated_at: row.get(12).map_err(|e| e.to_string())?, markdown_rel_path: row.get(13).map_err(|e| e.to_string())?, + taint, }); } Ok(docs) @@ -428,7 +443,7 @@ impl UnifiedMemory { if let Some(ns) = namespace { let mut stmt = conn .prepare( - "SELECT document_id, namespace, key, title, source_type, priority, created_at, updated_at + "SELECT document_id, namespace, key, title, source_type, priority, created_at, updated_at, taint FROM memory_docs WHERE namespace = ?1 ORDER BY updated_at DESC", ) .map_err(|e| format!("prepare list_documents: {e}"))?; @@ -448,12 +463,13 @@ impl UnifiedMemory { "priority": row.get::<_, String>(5).map_err(|e| e.to_string())?, "createdAt": row.get::<_, f64>(6).map_err(|e| e.to_string())?, "updatedAt": row.get::<_, f64>(7).map_err(|e| e.to_string())?, + "taint": row.get::<_, String>(8).map_err(|e| e.to_string())?, })); } } else { let mut stmt = conn .prepare( - "SELECT document_id, namespace, key, title, source_type, priority, created_at, updated_at + "SELECT document_id, namespace, key, title, source_type, priority, created_at, updated_at, taint FROM memory_docs ORDER BY updated_at DESC", ) .map_err(|e| format!("prepare list_documents: {e}"))?; @@ -473,6 +489,7 @@ impl UnifiedMemory { "priority": row.get::<_, String>(5).map_err(|e| e.to_string())?, "createdAt": row.get::<_, f64>(6).map_err(|e| e.to_string())?, "updatedAt": row.get::<_, f64>(7).map_err(|e| e.to_string())?, + "taint": row.get::<_, String>(8).map_err(|e| e.to_string())?, })); } } diff --git a/src/openhuman/memory_store/unified/documents_tests.rs b/src/openhuman/memory_store/unified/documents_tests.rs index 6e9936ad5..613b44267 100644 --- a/src/openhuman/memory_store/unified/documents_tests.rs +++ b/src/openhuman/memory_store/unified/documents_tests.rs @@ -26,6 +26,7 @@ fn make_doc_input( category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, } } @@ -939,6 +940,7 @@ async fn upsert_document_redacts_secret_like_content_before_persisting() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .unwrap(); @@ -1048,6 +1050,7 @@ async fn upsert_document_rejects_secret_like_key() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("secret-like key should be rejected"); @@ -1072,6 +1075,7 @@ async fn upsert_document_rejects_secret_like_namespace() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("secret-like namespace should be rejected"); @@ -1096,6 +1100,7 @@ async fn upsert_document_metadata_only_rejects_secret_like_key() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("secret-like key should be rejected"); @@ -1175,6 +1180,7 @@ async fn upsert_document_rejects_pii_like_key() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("PII-like key should be rejected"); @@ -1202,6 +1208,7 @@ async fn upsert_document_rejects_pii_like_namespace() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("PII-like namespace should be rejected"); @@ -1229,6 +1236,7 @@ async fn upsert_document_metadata_only_rejects_pii_like_key() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("PII-like key should be rejected"); @@ -1256,6 +1264,7 @@ async fn upsert_document_metadata_only_rejects_pii_like_namespace() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .expect_err("PII-like namespace should be rejected"); diff --git a/src/openhuman/memory_store/unified/init.rs b/src/openhuman/memory_store/unified/init.rs index 51b2d4d99..82c48ed81 100644 --- a/src/openhuman/memory_store/unified/init.rs +++ b/src/openhuman/memory_store/unified/init.rs @@ -99,6 +99,7 @@ impl UnifiedMemory { created_at REAL NOT NULL, updated_at REAL NOT NULL, markdown_rel_path TEXT NOT NULL, + taint TEXT NOT NULL DEFAULT 'internal', UNIQUE(namespace, key) ); CREATE INDEX IF NOT EXISTS idx_memory_docs_ns_updated ON memory_docs(namespace, updated_at DESC); @@ -173,6 +174,21 @@ impl UnifiedMemory { } } + // Backfill the `taint` column on existing `memory_docs` databases. + // Fresh installs get this via the CREATE TABLE above; older DBs need + // the ALTER so retrieval can carry the provenance signal up to the + // subconscious gate. Idempotent: a duplicate-column error on + // re-application is expected (logged at trace). + match conn.execute( + "ALTER TABLE memory_docs ADD COLUMN taint TEXT NOT NULL DEFAULT 'internal'", + [], + ) { + Ok(_) => tracing::debug!("[memory_docs:init] applied taint column migration"), + Err(e) => { + tracing::trace!("[memory_docs:init] taint column already present: {e}") + } + } + // Create FTS5 episodic tables (episodic_log, episodic_fts, and their // triggers) so the Archivist can call episodic_insert immediately after // the store is initialised. diff --git a/src/openhuman/memory_store/unified/query.rs b/src/openhuman/memory_store/unified/query.rs index 5164e249b..23e2c7812 100644 --- a/src/openhuman/memory_store/unified/query.rs +++ b/src/openhuman/memory_store/unified/query.rs @@ -92,6 +92,7 @@ impl UnifiedMemory { content: hit.content, score: hit.score, category: hit.category, + taint: hit.taint, }); } Ok(out) @@ -171,6 +172,7 @@ impl UnifiedMemory { document_id: Some(doc.document_id.clone()), chunk_id: best_chunk_id, supporting_relations, + taint: doc.taint, }); } @@ -209,6 +211,10 @@ impl UnifiedMemory { document_id: None, chunk_id: None, supporting_relations: Vec::new(), + // KV rows have no provenance column; conservatively + // surface as Internal so the subconscious gate doesn't + // mis-escalate user-state writes. + taint: crate::openhuman::memory::MemoryTaint::Internal, }); } @@ -287,6 +293,10 @@ impl UnifiedMemory { document_id: None, chunk_id: None, supporting_relations: Vec::new(), + // Episodic rows are derived from user chat turns and + // never carry sync-ingest content; surface as + // Internal so the subconscious gate trusts them. + taint: crate::openhuman::memory::MemoryTaint::Internal, }); } } @@ -325,6 +335,10 @@ impl UnifiedMemory { document_id: None, chunk_id: None, supporting_relations: Vec::new(), + // Event extractions are derived from chat segments; + // treat them as Internal until a future migration + // surfaces per-event provenance. + taint: crate::openhuman::memory::MemoryTaint::Internal, }); } @@ -428,6 +442,7 @@ impl UnifiedMemory { .map(|relation| RelationMatch { relation, hop: 1 }) .collect::>(), ), + taint: doc.taint, }); } @@ -462,6 +477,7 @@ impl UnifiedMemory { document_id: None, chunk_id: None, supporting_relations: Vec::new(), + taint: crate::openhuman::memory::MemoryTaint::Internal, }); } diff --git a/src/openhuman/memory_store/unified/query_tests.rs b/src/openhuman/memory_store/unified/query_tests.rs index afc9bbeb5..81e1164a7 100644 --- a/src/openhuman/memory_store/unified/query_tests.rs +++ b/src/openhuman/memory_store/unified/query_tests.rs @@ -62,6 +62,7 @@ async fn query_namespace_uses_graph_signal_for_document_ranking() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .unwrap(); @@ -104,6 +105,7 @@ async fn query_scores_relation_entities_found_in_document_content() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .unwrap(); @@ -280,6 +282,7 @@ async fn query_supporting_relations_contain_entity_types() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .unwrap(); @@ -386,6 +389,7 @@ async fn format_context_text_includes_entity_types() { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, }) .await .unwrap(); @@ -467,6 +471,7 @@ fn pref_doc(key: &str, content: &str) -> NamespaceDocumentInput { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, } } @@ -633,6 +638,7 @@ fn situational_doc(key: &str, content: &str) -> NamespaceDocumentInput { category: "core".to_string(), session_id: None, document_id: None, + taint: crate::openhuman::memory::MemoryTaint::Internal, } } diff --git a/src/openhuman/memory_tools/test_helpers.rs b/src/openhuman/memory_tools/test_helpers.rs index b74690465..b39e06127 100644 --- a/src/openhuman/memory_tools/test_helpers.rs +++ b/src/openhuman/memory_tools/test_helpers.rs @@ -42,6 +42,7 @@ impl Memory for MockMemory { timestamp: "now".into(), session_id: session_id.map(str::to_string), score: None, + taint: Default::default(), }, ); Ok(()) diff --git a/src/openhuman/screen_intelligence/helpers.rs b/src/openhuman/screen_intelligence/helpers.rs index 33b45ba61..12960408f 100644 --- a/src/openhuman/screen_intelligence/helpers.rs +++ b/src/openhuman/screen_intelligence/helpers.rs @@ -208,6 +208,10 @@ pub(crate) async fn persist_vision_summary( category: VISION_MEMORY_CATEGORY.to_string(), session_id: None, document_id: None, + // Screen-intelligence captures the user's own active window; + // not third-party sync content. Internal preserves the + // baseline subconscious-trust behaviour. + taint: crate::openhuman::memory::MemoryTaint::Internal, }; // put_doc_light stores the document (DB row + markdown file) without diff --git a/src/openhuman/security/policy.rs b/src/openhuman/security/policy.rs index 53235d926..b620b07d1 100644 --- a/src/openhuman/security/policy.rs +++ b/src/openhuman/security/policy.rs @@ -361,8 +361,8 @@ impl Default for SecurityPolicy { mod policy_command; use policy_command::{ classify_segment, command_basename, contains_unquoted_char, contains_unquoted_single_ampersand, - has_dangerous_env_prefix, has_hidden_execution, is_command_executor, normalized_command_name, - skip_env_assignments, split_unquoted_segments, + has_dangerous_env_prefix, has_hidden_execution, has_leading_env_assignment, + is_command_executor, normalized_command_name, skip_env_assignments, split_unquoted_segments, }; impl SecurityPolicy { @@ -714,12 +714,20 @@ impl SecurityPolicy { // Split on unquoted command separators and validate each sub-command. let segments = split_unquoted_segments(command); for segment in &segments { - // Reject segments that prefix the command with a dangerous env - // assignment (e.g. `GIT_PAGER= git log`). The bare command - // after the assignment is allowlisted, but the prefix mutates - // the downstream binary's execution to spawn `` as a - // subprocess. See [`has_dangerous_env_prefix`]. - if has_dangerous_env_prefix(segment) { + // Reject ANY segment that prefixes the command with an env-var + // assignment, not just the known-dangerous names. Helper-style + // exec primitives (`GIT_SSH=./wrapper git ls-remote`, + // `SSH_ASKPASS=./prompt ssh user@host`, `LD_PRELOAD=./libx.so + // ls`, etc.) change which binary the allowed command actually + // resolves to — or change its behaviour via a hook — without + // any blocked command name ever appearing in the segment. The + // allowlist already names every command we want to permit, and + // none of those commands need an operator-set env var at + // invoke time, so the broader gate has no false-positive + // surface on the approved path. `has_dangerous_env_prefix` + // remains in the source for legacy tests and as the + // narrower-grained signal. + if has_leading_env_assignment(segment) || has_dangerous_env_prefix(segment) { return false; } diff --git a/src/openhuman/security/policy_command.rs b/src/openhuman/security/policy_command.rs index f255652a9..996bf367f 100644 --- a/src/openhuman/security/policy_command.rs +++ b/src/openhuman/security/policy_command.rs @@ -55,6 +55,12 @@ const DANGEROUS_ENV_PREFIXES: &[&str] = &[ /// Returns true if `s` starts with one or more inline env assignments and any /// of the assigned names are in [`DANGEROUS_ENV_PREFIXES`]. /// +/// Now superseded by [`has_leading_env_assignment`] for the +/// allowlist check (which rejects ANY leading env assignment), but kept +/// for callers that specifically want the dangerous-only signal — +/// notably tests that pin the old DANGEROUS_ENV_PREFIXES rejection +/// shape. +/// /// The allowlist validation in [`SecurityPolicy::is_command_allowed`] uses /// [`skip_env_assignments`] to look past the env prefix before matching the /// command name. That leaves a class of attacks where the bare command (e.g. @@ -92,6 +98,44 @@ pub(super) fn has_dangerous_env_prefix(s: &str) -> bool { } } +/// Returns true if `s` starts with at least one inline env-var +/// assignment of the shape `NAME=...`, where `NAME` begins with an ASCII +/// letter or underscore. Catches the entire `GIT_SSH=…`, `SSH_ASKPASS=…`, +/// `LD_PRELOAD=…`, `IFS=…` family — including names we haven't enumerated +/// in [`DANGEROUS_ENV_PREFIXES`] — by treating ANY leading assignment as +/// suspect. The allowlist already names every command we want to permit; +/// nothing in that list needs the operator to set an env var at invoke +/// time, so the broader gate has no false-positive surface on the +/// approved path. +/// +/// Used by [`SecurityPolicy::is_command_allowed`] as a structural guard +/// alongside the existing dangerous-prefix check. +pub(super) fn has_leading_env_assignment(s: &str) -> bool { + let Some(word) = s.split_whitespace().next() else { + return false; + }; + let Some((name, _value)) = word.split_once('=') else { + return false; + }; + if name.is_empty() { + return false; + } + // Identifier shape: first char letter or `_`, the rest alphanumeric + // or `_`. Anything else (e.g. `foo[bar]=`) is not a shell assignment. + let mut chars = name.chars(); + let first = match chars.next() { + Some(c) => c, + None => return false, + }; + if !(first.is_ascii_alphabetic() || first == '_') { + return false; + } + if !chars.all(|c| c.is_ascii_alphanumeric() || c == '_') { + return false; + } + true +} + /// Skip leading environment variable assignments (e.g. `FOO=bar cmd args`). /// Returns the remainder starting at the first non-assignment word. pub(super) fn skip_env_assignments(s: &str) -> &str { diff --git a/src/openhuman/security/policy_tests.rs b/src/openhuman/security/policy_tests.rs index 3769ddfa0..7ae871e6c 100644 --- a/src/openhuman/security/policy_tests.rs +++ b/src/openhuman/security/policy_tests.rs @@ -1274,12 +1274,15 @@ fn dangerous_env_var_prefix_blocked() { // Case-insensitive: should also catch mixed-case names. assert!(!p.is_command_allowed("Ld_PrElOaD=/tmp/x.so git status")); - // Benign env prefixes still pass: TZ, LANG, LC_ALL, custom names that - // don't trigger downstream subprocess execution. - assert!(p.is_command_allowed("TZ=UTC git log")); - assert!(p.is_command_allowed("LANG=en_US.UTF-8 git log")); - assert!(p.is_command_allowed("LC_ALL=C git status")); - assert!(p.is_command_allowed("FOO=bar git status")); + // All leading env-var assignments are now rejected — including + // previously-benign-looking ones (TZ, LANG, LC_ALL, custom names). + // The allowlist already names every command we want to permit, and + // none need an operator-set env var at invoke time, so the broader + // gate has no false-positive surface on the approved path. + assert!(!p.is_command_allowed("TZ=UTC git log")); + assert!(!p.is_command_allowed("LANG=en_US.UTF-8 git log")); + assert!(!p.is_command_allowed("LC_ALL=C git status")); + assert!(!p.is_command_allowed("FOO=bar git status")); // No env prefix at all — unchanged. assert!(p.is_command_allowed("git status")); } @@ -1351,15 +1354,33 @@ fn command_injection_process_substitution_blocked() { } #[test] -fn command_env_var_prefix_with_allowed_cmd() { +fn command_env_var_prefix_is_always_rejected() { let p = default_policy(); - // env assignment + allowed command — OK - assert!(p.is_command_allowed("FOO=bar ls")); - assert!(p.is_command_allowed("LANG=C grep pattern file")); - // env assignment + disallowed command — blocked + // ANY env assignment is rejected — including in front of an + // otherwise-allowed command. Helper-style exec primitives + // (GIT_SSH=, SSH_ASKPASS=, LD_PRELOAD=) and benign-looking + // overrides (FOO=, LANG=) both go through the same gate so the + // policy doesn't have to enumerate every shape of every + // downstream tool's hook surface. + assert!(!p.is_command_allowed("FOO=bar ls")); + assert!(!p.is_command_allowed("LANG=C grep pattern file")); assert!(!p.is_command_allowed("FOO=bar rm -rf /")); } +#[test] +fn validate_command_rejects_leading_env_var_assignment() { + let p = default_policy(); + // Helper-style exec primitives that mutate which binary the + // approved command actually runs as: rejected. + assert!(!p.is_command_allowed("GIT_SSH=./wrapper.sh git ls-remote ssh://x")); + assert!(!p.is_command_allowed("SSH_ASKPASS=./y ssh user@host")); + assert!(!p.is_command_allowed("LD_PRELOAD=./libx.so ls")); + // Negative: same command without the env prefix passes the + // structural guard (it may still fail later on its own merits, + // but the env-prefix gate doesn't fire). + assert!(p.is_command_allowed("git ls-remote ssh://example.com")); +} + // -- Edge cases: path traversal ----------------------------------- #[test] diff --git a/src/openhuman/skills/schemas.rs b/src/openhuman/skills/schemas.rs index 20eb96d5e..3024048d4 100644 --- a/src/openhuman/skills/schemas.rs +++ b/src/openhuman/skills/schemas.rs @@ -920,11 +920,14 @@ pub(crate) async fn spawn_skill_run_background( // toolchain are loaded fresh per run; the parent returns the handle // immediately. Same flow handle_skills_run used to inline — extracted // so the `run_skill` agent tool can re-use it for skill chaining. + let inherited_origin = crate::openhuman::agent::turn_origin::current() + .unwrap_or(crate::openhuman::agent::turn_origin::AgentTurnOrigin::Cli); { let run_id = run_id.clone(); let skill_id = skill_id.clone(); let inputs = inputs.clone(); let log_path = log_path.clone(); + let inherited_origin = inherited_origin.clone(); tokio::spawn(async move { if let Err(e) = run_log::write_header(&log_path, &skill_id, &run_id, &inputs, &task_prompt).await @@ -974,9 +977,15 @@ pub(crate) async fn spawn_skill_run_background( let bridge = tokio::spawn(run_log::drain_to_log(rx, log_path.clone())); let started = std::time::Instant::now(); - let result = - with_autonomous_iter_cap(SKILL_RUN_MAX_ITERATIONS, agent.run_single(&task_prompt)) - .await; + // Inherit the parent turn's origin so a skill triggered from an + // ExternalChannel / tainted context retains its provenance + // through the approval gate. Falls back to Cli for direct + // user-initiated RPC / CLI flows. + let result = crate::openhuman::agent::turn_origin::with_origin( + inherited_origin, + with_autonomous_iter_cap(SKILL_RUN_MAX_ITERATIONS, agent.run_single(&task_prompt)), + ) + .await; agent.set_on_progress(None); drop(agent); let _ = bridge.await; diff --git a/src/openhuman/subconscious/engine.rs b/src/openhuman/subconscious/engine.rs index 73c42c43c..ee4e3abf7 100644 --- a/src/openhuman/subconscious/engine.rs +++ b/src/openhuman/subconscious/engine.rs @@ -21,6 +21,27 @@ use std::sync::Arc; use tokio::sync::Mutex; use tracing::{debug, info, warn}; +/// Pick the `TrustedAutomationSource` variant for a subconscious tick. +/// +/// Extracted from the engine's `run_agent` body so the +/// origin-escalation contract can be unit-tested without spinning up +/// a real `Agent` + provider. +/// +/// Contract: any tick whose situation report contained third-party +/// sync content (Gmail / Slack / Notion / sealed source summaries) +/// must run with `SubconsciousTainted` so the approval gate refuses +/// external_effect tools. Untainted ticks keep the legacy +/// `Subconscious` origin. +pub(crate) fn tick_origin_source( + has_external_content: bool, +) -> crate::openhuman::agent::turn_origin::TrustedAutomationSource { + if has_external_content { + crate::openhuman::agent::turn_origin::TrustedAutomationSource::SubconsciousTainted + } else { + crate::openhuman::agent::turn_origin::TrustedAutomationSource::Subconscious + } +} + pub struct SubconsciousEngine { workspace_dir: PathBuf, mode: SubconsciousMode, @@ -167,13 +188,16 @@ impl SubconsciousEngine { &recent_reflections, ) .await; + let has_external_content = report.has_external_content; // 2. Load identity context let identity = prompt::load_identity_context(&self.workspace_dir); // 3. Run the subconscious agent - let agent_prompt = prompt::build_agent_prompt(&report, &identity); - let agent_result = self.run_agent(&config, &agent_prompt).await; + let agent_prompt = prompt::build_agent_prompt(&report.prompt_text, &identity); + let agent_result = self + .run_agent(&config, &agent_prompt, has_external_content) + .await; let agent_failed = agent_result.is_err(); let drafts = agent_result.unwrap_or_default(); @@ -258,6 +282,7 @@ impl SubconsciousEngine { &self, config: &Config, prompt_text: &str, + has_external_content: bool, ) -> Result, String> { use crate::openhuman::agent::Agent; @@ -296,7 +321,29 @@ impl SubconsciousEngine { ); debug!("[subconscious] spawning agent with tool access"); - let response = agent.run_single(&user_message).await.map_err(|e| { + // Subconscious ticks are trusted automation: the user enabled the + // background loop knowing it should think about their state. + // When the situation report carries content derived from third- + // party sync sources (Gmail / Slack / Notion / sealed source + // summaries), escalate the origin so the approval gate refuses + // external_effect tools for the rest of the tick — a hostile + // upstream message can otherwise nudge the LLM into a tool call + // the user would never have authorised. + let source = tick_origin_source(has_external_content); + debug!( + "[subconscious] tick origin source={:?} has_external_content={has_external_content}", + source + ); + let origin = crate::openhuman::agent::turn_origin::AgentTurnOrigin::TrustedAutomation { + job_id: format!("subconscious:tick:{}", now_secs() as u64), + source, + }; + let response = crate::openhuman::agent::turn_origin::with_origin( + origin, + agent.run_single(&user_message), + ) + .await + .map_err(|e| { warn!("[subconscious] agent run failed: {e}"); format!("agent run: {e}") })?; diff --git a/src/openhuman/subconscious/engine_tests.rs b/src/openhuman/subconscious/engine_tests.rs index 6ff315f1c..71bb38c3b 100644 --- a/src/openhuman/subconscious/engine_tests.rs +++ b/src/openhuman/subconscious/engine_tests.rs @@ -57,3 +57,22 @@ fn extract_json_finds_array() { assert!(extracted.starts_with('[')); assert!(extracted.ends_with(']')); } + +// ── Tick origin upgrade (#approval-origin) ────────────────────────────── + +#[test] +fn tick_origin_untainted_keeps_subconscious_source() { + use crate::openhuman::agent::turn_origin::TrustedAutomationSource; + let source = tick_origin_source(false); + assert!(matches!(source, TrustedAutomationSource::Subconscious)); +} + +#[test] +fn tick_origin_with_external_sync_chunk_uses_tainted_source() { + use crate::openhuman::agent::turn_origin::TrustedAutomationSource; + let source = tick_origin_source(true); + assert!(matches!( + source, + TrustedAutomationSource::SubconsciousTainted + )); +} diff --git a/src/openhuman/subconscious/situation_report/mod.rs b/src/openhuman/subconscious/situation_report/mod.rs index 61eeadf30..e98668ee4 100644 --- a/src/openhuman/subconscious/situation_report/mod.rs +++ b/src/openhuman/subconscious/situation_report/mod.rs @@ -40,6 +40,20 @@ mod summaries; /// Rough chars-per-token estimate for budget enforcement. const CHARS_PER_TOKEN: usize = 4; +/// Result of building a subconscious situation report. +/// +/// `has_external_content` is true iff the prompt now contains content +/// derived from third-party sync sources (Gmail / Slack / Notion / chat +/// transcripts / sealed source summaries). The subconscious engine uses +/// this signal to upgrade the tick's `AgentTurnOrigin` to +/// `TrustedAutomationSource::SubconsciousTainted`, which makes the +/// approval gate refuse external_effect tools for the rest of the tick. +#[derive(Debug, Clone)] +pub struct SituationReport { + pub prompt_text: String, + pub has_external_content: bool, +} + /// Build the situation report for one subconscious tick. /// /// `last_tick_at` is 0.0 on cold start (include everything in the @@ -54,10 +68,11 @@ pub async fn build_situation_report( last_tick_at: f64, token_budget: u32, recent_reflections: &[Reflection], -) -> String { +) -> SituationReport { let char_budget = (token_budget as usize) * CHARS_PER_TOKEN; let mut report = String::with_capacity(char_budget.min(64_000)); let mut remaining = char_budget; + let mut has_external_content = false; // Section 1: environment anchor. let env_section = build_environment_section(workspace_dir); @@ -74,12 +89,15 @@ pub async fn build_situation_report( append_section(&mut report, &mut remaining, &tasks_section); // Section 4: recently-sealed source summaries since last tick. - let summaries_section = summaries::build_section(config, last_tick_at).await; + let (summaries_section, summaries_tainted) = + summaries::build_section(config, last_tick_at).await; append_section(&mut report, &mut remaining, &summaries_section); + has_external_content |= summaries_tainted; // Section 5: source-tree recap window since last tick. - let recap_section = query_window::build_section(config, last_tick_at).await; + let (recap_section, recap_tainted) = query_window::build_section(config, last_tick_at).await; append_section(&mut report, &mut remaining, &recap_section); + has_external_content |= recap_tainted; // Section 6: previous reflections (anti-double-emit context). let reflections_section = reflections::build_section(recent_reflections); @@ -89,7 +107,10 @@ pub async fn build_situation_report( report.push_str("No state changes detected since last tick.\n"); } - report + SituationReport { + prompt_text: report, + has_external_content, + } } fn build_environment_section(workspace_dir: &Path) -> String { diff --git a/src/openhuman/subconscious/situation_report/query_window.rs b/src/openhuman/subconscious/situation_report/query_window.rs index ff17f3a9a..71431c2ce 100644 --- a/src/openhuman/subconscious/situation_report/query_window.rs +++ b/src/openhuman/subconscious/situation_report/query_window.rs @@ -24,7 +24,14 @@ const MIN_WINDOW_DAYS: u32 = 1; /// Max source summaries to pull into the recap window. const MAX_RECAP_HITS: usize = 20; -pub async fn build_section(config: &Config, last_tick_at: f64) -> String { +/// Build the source-tree recap window section. +/// +/// Returns `(section_markdown, has_external_content)` — the bool is +/// `true` iff at least one fresh source-tree hit was rendered, which +/// means the prompt now carries third-party sync content. See the +/// matching note on `summaries::build_section` for why the caller uses +/// this to upgrade the tick's turn origin. +pub async fn build_section(config: &Config, last_tick_at: f64) -> (String, bool) { let window_days = compute_window_days(last_tick_at); log::debug!( "[subconscious::situation_report::query_window] window_days={window_days} \ @@ -36,7 +43,7 @@ pub async fn build_section(config: &Config, last_tick_at: f64) -> String { Ok(r) => r, Err(e) => { log::warn!("[subconscious::situation_report::query_window] failed: {e}"); - return "## Recap window\n\nRecap unavailable.\n".to_string(); + return ("## Recap window\n\nRecap unavailable.\n".to_string(), false); } }; @@ -61,10 +68,13 @@ pub async fn build_section(config: &Config, last_tick_at: f64) -> String { }; if fresh_hits.is_empty() { - return format!( - "## Recap window ({} day{})\n\nNo new recap content since last tick.\n", - window_days, - if window_days == 1 { "" } else { "s" } + return ( + format!( + "## Recap window ({} day{})\n\nNo new recap content since last tick.\n", + window_days, + if window_days == 1 { "" } else { "s" } + ), + false, ); } @@ -83,7 +93,7 @@ pub async fn build_section(config: &Config, last_tick_at: f64) -> String { truncate(&hit.content, 600) ); } - section + (section, true) } fn compute_window_days(last_tick_at: f64) -> u32 { diff --git a/src/openhuman/subconscious/situation_report/summaries.rs b/src/openhuman/subconscious/situation_report/summaries.rs index 58919e605..af467989d 100644 --- a/src/openhuman/subconscious/situation_report/summaries.rs +++ b/src/openhuman/subconscious/situation_report/summaries.rs @@ -14,7 +14,16 @@ const MAX_SUMMARIES: usize = 8; /// Per-summary content cap — keep prompts compact. const SUMMARY_CONTENT_PREVIEW: usize = 320; -pub async fn build_section(config: &Config, last_tick_at: f64) -> String { +/// Build the recently-sealed-summaries section. +/// +/// Returns `(section_markdown, has_external_content)` — the bool is +/// `true` iff at least one summary row was actually rendered (i.e. the +/// section contains third-party sync content, not the empty +/// "No new sealed summaries" placeholder). The caller uses the flag to +/// upgrade the subconscious turn origin to +/// [`crate::openhuman::agent::turn_origin::TrustedAutomationSource::SubconsciousTainted`] +/// so external_effect tools are refused for the rest of the tick. +pub async fn build_section(config: &Config, last_tick_at: f64) -> (String, bool) { log::debug!( "[subconscious::situation_report::summaries] building section last_tick_at={last_tick_at}" ); @@ -30,12 +39,18 @@ pub async fn build_section(config: &Config, last_tick_at: f64) -> String { Ok(rows) => rows, Err(e) => { log::warn!("[subconscious::situation_report::summaries] read failed: {e}"); - return "## Recent summaries\n\nSummaries unavailable.\n".to_string(); + return ( + "## Recent summaries\n\nSummaries unavailable.\n".to_string(), + false, + ); } }; if rows.is_empty() { - return "## Recent summaries\n\nNo new sealed summaries since last tick.\n".to_string(); + return ( + "## Recent summaries\n\nNo new sealed summaries since last tick.\n".to_string(), + false, + ); } let mut section = String::from("## Recent summaries\n\n"); @@ -53,7 +68,7 @@ pub async fn build_section(config: &Config, last_tick_at: f64) -> String { row.tree_scope, row.level, row.summary_id, preview ); } - section + (section, true) } #[derive(Debug)] diff --git a/src/openhuman/tools/impl/system/tool_stats.rs b/src/openhuman/tools/impl/system/tool_stats.rs index 5a8711f36..76ac767f1 100644 --- a/src/openhuman/tools/impl/system/tool_stats.rs +++ b/src/openhuman/tools/impl/system/tool_stats.rs @@ -167,6 +167,7 @@ mod tests { timestamp: "now".into(), session_id: session_id.map(str::to_string), score: None, + taint: Default::default(), }, ); Ok(()) diff --git a/tests/agent_harness_leftovers_raw_coverage_e2e.rs b/tests/agent_harness_leftovers_raw_coverage_e2e.rs index 30b39e833..ee4368eb2 100644 --- a/tests/agent_harness_leftovers_raw_coverage_e2e.rs +++ b/tests/agent_harness_leftovers_raw_coverage_e2e.rs @@ -124,6 +124,7 @@ impl Memory for StubMemory { timestamp: "2026-05-29T00:00:00Z".to_string(), session_id: session_id.map(str::to_string), score: Some(0.9), + taint: Default::default(), }); Ok(()) } diff --git a/tests/agent_memory_loader_public.rs b/tests/agent_memory_loader_public.rs index 4e95ce1fb..ef429ad91 100644 --- a/tests/agent_memory_loader_public.rs +++ b/tests/agent_memory_loader_public.rs @@ -81,6 +81,7 @@ fn entry(key: &str, content: &str, score: Option) -> MemoryEntry { timestamp: "now".into(), session_id: None, score, + taint: Default::default(), } } diff --git a/tests/agent_session_round24_raw_coverage_e2e.rs b/tests/agent_session_round24_raw_coverage_e2e.rs index 7a2f8123e..b75b55079 100644 --- a/tests/agent_session_round24_raw_coverage_e2e.rs +++ b/tests/agent_session_round24_raw_coverage_e2e.rs @@ -187,6 +187,7 @@ impl Memory for RecordingMemory { timestamp: "2026-05-30T00:00:00Z".to_string(), session_id: None, score: Some(0.98), + taint: Default::default(), }] .into_iter() .take(limit) @@ -354,6 +355,7 @@ fn entry(key: &str, namespace: &str, content: &str) -> MemoryEntry { timestamp: "2026-05-30T00:00:00Z".to_string(), session_id: None, score: Some(0.9), + taint: Default::default(), } } diff --git a/tests/agent_session_turn_raw_coverage_e2e.rs b/tests/agent_session_turn_raw_coverage_e2e.rs index 48be5ca89..874084580 100644 --- a/tests/agent_session_turn_raw_coverage_e2e.rs +++ b/tests/agent_session_turn_raw_coverage_e2e.rs @@ -190,6 +190,7 @@ impl Memory for StaticMemory { timestamp: "2026-05-29T00:00:00Z".to_string(), session_id: session_id.map(str::to_string), score: Some(0.95), + taint: Default::default(), }); Ok(()) } @@ -755,6 +756,7 @@ async fn turn_xml_failures_checkpoint_policy_visibility_and_hooks_are_publicly_e timestamp: "2026-05-29T00:00:00Z".to_string(), session_id: None, score: Some(0.9), + taint: Default::default(), }]), fail_recall: true, })) diff --git a/tests/agent_tool_loop_raw_coverage_e2e.rs b/tests/agent_tool_loop_raw_coverage_e2e.rs index 51d88bc00..4c2739ba4 100644 --- a/tests/agent_tool_loop_raw_coverage_e2e.rs +++ b/tests/agent_tool_loop_raw_coverage_e2e.rs @@ -235,6 +235,7 @@ impl Memory for NoopMemory { timestamp: "2026-05-29T00:00:00Z".to_string(), session_id: session_id.map(str::to_string), score: Some(1.0), + taint: Default::default(), }); Ok(()) } @@ -379,6 +380,7 @@ async fn run_bus_turn( visible_tool_names, extra_tools: Vec::new(), on_progress: None, + origin: openhuman_core::openhuman::agent::turn_origin::AgentTurnOrigin::Cli, }, ) .await diff --git a/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs b/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs index bf08435db..3499e7ebd 100644 --- a/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs +++ b/tests/agent_turn_builder_leftovers_raw_coverage_e2e.rs @@ -167,6 +167,7 @@ impl Memory for RecordingMemory { timestamp: "2026-05-30T00:00:00Z".to_string(), session_id: None, score: Some(0.91), + taint: Default::default(), }] .into_iter() .take(limit) diff --git a/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs b/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs index f9128acf2..0da3290ae 100644 --- a/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs +++ b/tests/agent_turn_toolloop_round22_raw_coverage_e2e.rs @@ -225,6 +225,7 @@ async fn run_turn( visible_tool_names: None, extra_tools: Vec::new(), on_progress, + origin: openhuman_core::openhuman::agent::turn_origin::AgentTurnOrigin::Cli, }, ) .await diff --git a/tests/inference_agent_raw_coverage_e2e.rs b/tests/inference_agent_raw_coverage_e2e.rs index 18216ee93..b46562280 100644 --- a/tests/inference_agent_raw_coverage_e2e.rs +++ b/tests/inference_agent_raw_coverage_e2e.rs @@ -567,6 +567,7 @@ fn memory_entry( timestamp: "2026-05-29T12:00:00Z".to_string(), session_id: session_id.map(ToOwned::to_owned), score, + taint: Default::default(), } } @@ -2866,6 +2867,7 @@ async fn agent_triage_evaluator_covers_native_dispatch_decision_and_deferred_pat visible_tool_names: Some(HashSet::new()), extra_tools: Vec::new(), on_progress: None, + origin: openhuman_core::openhuman::agent::turn_origin::AgentTurnOrigin::Cli, }, ) .await diff --git a/tests/memory_graph_sync_e2e.rs b/tests/memory_graph_sync_e2e.rs index e76fc6dd3..1ab8adeca 100644 --- a/tests/memory_graph_sync_e2e.rs +++ b/tests/memory_graph_sync_e2e.rs @@ -74,6 +74,7 @@ async fn ingest_document_populates_namespace_graph() { category: "core".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }, config: ci_safe_config(), }) @@ -198,6 +199,7 @@ async fn put_doc_background_extraction_then_graph_query() { category: "core".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("put_doc"); diff --git a/tests/memory_raw_coverage_e2e.rs b/tests/memory_raw_coverage_e2e.rs index 146daa5be..c2480f0c7 100644 --- a/tests/memory_raw_coverage_e2e.rs +++ b/tests/memory_raw_coverage_e2e.rs @@ -561,6 +561,7 @@ async fn memory_ingestion_state_and_request_models_report_edges() { category: "core".into(), session_id: Some("session-1".into()), document_id: Some("doc-1".into()), + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }, config: cfg.clone(), }; diff --git a/tests/memory_threads_raw_coverage_e2e.rs b/tests/memory_threads_raw_coverage_e2e.rs index 681e8eccd..406fad5c9 100644 --- a/tests/memory_threads_raw_coverage_e2e.rs +++ b/tests/memory_threads_raw_coverage_e2e.rs @@ -466,6 +466,7 @@ Kitchen is north of Garden. category: "core".into(), session_id: Some("session-coverage".into()), document_id: Some("doc-memory-raw-ingestion".into()), + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }, config: MemoryIngestionConfig::default(), }) @@ -535,6 +536,7 @@ Kitchen is north of Garden. category: "core".into(), session_id: None, document_id: Some("doc-memory-raw-ingestion".into()), + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }, &MemoryIngestionConfig { extraction_mode: openhuman_core::openhuman::memory::ExtractionMode::Chunk, @@ -2056,6 +2058,7 @@ fn memory_retrieval_embedding_and_rpc_model_helpers_round_trip() { timestamp: now.to_rfc3339(), session_id: Some("session-1".into()), score: Some(0.9), + taint: Default::default(), }; assert_eq!(entry.category.to_string(), "testing"); let opts = RecallOpts { diff --git a/tests/memory_tree_memory_round23_raw_coverage_e2e.rs b/tests/memory_tree_memory_round23_raw_coverage_e2e.rs index c4fd639b8..3da625f01 100644 --- a/tests/memory_tree_memory_round23_raw_coverage_e2e.rs +++ b/tests/memory_tree_memory_round23_raw_coverage_e2e.rs @@ -133,6 +133,7 @@ Bob Builder will review the memory tree recap. category: "coverage".into(), session_id: Some("round23-session".into()), document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }, config: MemoryIngestionConfig { model_name: "round23-heuristic".into(), diff --git a/tests/personality_e2e.rs b/tests/personality_e2e.rs index 6b7c1cb31..e96cc80de 100644 --- a/tests/personality_e2e.rs +++ b/tests/personality_e2e.rs @@ -265,6 +265,7 @@ async fn two_personalities_have_isolated_sqlite_stores() { category: "core".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("write default"); @@ -282,6 +283,7 @@ async fn two_personalities_have_isolated_sqlite_stores() { category: "core".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("write alice"); @@ -337,6 +339,7 @@ async fn personality_memory_persists_across_reopens() { category: "core".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("write"); diff --git a/tests/screen_intelligence_vision_e2e.rs b/tests/screen_intelligence_vision_e2e.rs index 9728f15e1..e22491a38 100644 --- a/tests/screen_intelligence_vision_e2e.rs +++ b/tests/screen_intelligence_vision_e2e.rs @@ -286,6 +286,7 @@ async fn vision_pipeline_compress_parse_persist() { category: "screen_intelligence".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("upsert_document"); @@ -360,6 +361,7 @@ async fn multiple_vision_summaries_persist_and_query() { category: "screen_intelligence".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("upsert"); @@ -472,6 +474,7 @@ async fn vision_summary_upsert_is_idempotent() { category: "screen_intelligence".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("first upsert"); @@ -489,6 +492,7 @@ async fn vision_summary_upsert_is_idempotent() { category: "screen_intelligence".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("second upsert"); @@ -677,6 +681,7 @@ async fn vision_summary_struct_persist_and_deserialize_roundtrip() { category: "screen_intelligence".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }) .await .expect("upsert_document"); diff --git a/tests/subconscious_e2e.rs b/tests/subconscious_e2e.rs index 2e9937a56..d1774a777 100644 --- a/tests/subconscious_e2e.rs +++ b/tests/subconscious_e2e.rs @@ -33,6 +33,7 @@ async fn ingest_doc( category: "core".to_string(), session_id: None, document_id: None, + taint: openhuman_core::openhuman::memory::MemoryTaint::Internal, }, config: ci_safe_ingestion_config(), }) diff --git a/tests/tool_registry_approval_raw_coverage_e2e.rs b/tests/tool_registry_approval_raw_coverage_e2e.rs index 1f1775a13..5e9acd5f3 100644 --- a/tests/tool_registry_approval_raw_coverage_e2e.rs +++ b/tests/tool_registry_approval_raw_coverage_e2e.rs @@ -17,6 +17,7 @@ use tempfile::{tempdir, TempDir}; use openhuman_core::core::auth::{init_rpc_token, CORE_TOKEN_ENV_VAR}; use openhuman_core::core::jsonrpc::build_core_http_router; +use openhuman_core::openhuman::agent::turn_origin::{self, AgentTurnOrigin}; use openhuman_core::openhuman::approval::gate::{ parse_approval_reply, ApprovalChatContext, ApprovalGate, APPROVAL_CHAT_CONTEXT, }; @@ -667,6 +668,11 @@ async fn tool_registry_entries_include_connected_mcp_client_tools() { #[tokio::test] async fn tool_registry_schema_handlers_validate_and_return_payloads() { + // Acquire the env lock — this test loads Config via the diagnostics + // handler, and a sibling test temporarily points OPENHUMAN_WORKSPACE at + // a file to exercise the load-failure branch. Without the lock those + // two can race and this test sees the corrupted env. + let _lock = env_lock(); let schemas = all_tool_registry_controller_schemas(); assert_eq!( schemas @@ -1167,8 +1173,14 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { let gate_for_task = gate.clone(); let approval_task = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + // Scope a WebChat origin alongside the chat context — the gate now + // requires an origin label or it fails closed on `Unknown`. + turn_origin::with_origin( + AgentTurnOrigin::WebChat { + thread_id: "approval-raw-thread".to_string(), + client_id: "approval-raw-client".to_string(), + }, + APPROVAL_CHAT_CONTEXT.scope( ApprovalChatContext { thread_id: "approval-raw-thread".to_string(), client_id: "approval-raw-client".to_string(), @@ -1186,8 +1198,9 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { ) .await }, - ) - .await + ), + ) + .await }); let deadline = Instant::now() + Duration::from_secs(5); @@ -1301,6 +1314,11 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { "approve_always_for_tool should persist an auto-approve entry" ); + // Bare call with neither a chat context nor an AgentTurnOrigin scope: + // the gate now treats this as `Unknown` and fails closed (refuses to + // execute an external_effect tool from an unlabelled call site). The + // earlier "non-chat ⇒ Allow" behaviour leaked trusted execution to any + // caller that forgot to scope a label. let no_chat = gate .intercept_audited( "tools.web_search", @@ -1308,13 +1326,18 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { json!({ "query": "coverage" }), ) .await; - assert!(matches!( - no_chat.0, - openhuman_core::openhuman::approval::GateOutcome::Allow - )); + match &no_chat.0 { + openhuman_core::openhuman::approval::GateOutcome::Deny { reason } => { + assert!( + reason.contains("no origin label"), + "unlabelled call should be denied for missing origin: {reason}" + ); + } + other => panic!("expected Deny for unlabelled call, got {other:?}"), + } assert_eq!( no_chat.1, None, - "non-chat calls should not create approval rows" + "denied calls should not create approval rows" ); assert!(matches!( gate.intercept( @@ -1323,16 +1346,35 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { json!({ "query": "legacy" }), ) .await, - GateOutcome::Allow + GateOutcome::Deny { .. } )); - let auto_approved = gate - .intercept_audited( + // Always-allowed tools should bypass approval even when an origin is + // scoped — the auto_approve allowlist short-circuit runs before the + // origin branch. Install a live policy with the persisted entry so the + // gate sees the latest auto_approve set (the gate's boot-time config + // snapshot predates the approve_always_for_tool decision we just made). + live_policy::install( + Arc::new(SecurityPolicy { + workspace_dir: config.workspace_dir.clone(), + auto_approve: vec!["tools.composio_execute".to_string()], + ..SecurityPolicy::default() + }), + config.workspace_dir.clone(), + config.workspace_dir.clone(), + ); + let auto_approved = turn_origin::with_origin( + AgentTurnOrigin::WebChat { + thread_id: "approval-auto-thread".to_string(), + client_id: "approval-auto-client".to_string(), + }, + gate.intercept_audited( "tools.composio_execute", "tools.composio_execute(action=execute)", json!({ "action": "execute" }), - ) - .await; + ), + ) + .await; assert!(matches!( auto_approved.0, openhuman_core::openhuman::approval::GateOutcome::Allow @@ -1372,8 +1414,12 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { let gate_for_deny_task = gate.clone(); let deny_task = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + turn_origin::with_origin( + AgentTurnOrigin::WebChat { + thread_id: "approval-deny-thread".to_string(), + client_id: "approval-deny-client".to_string(), + }, + APPROVAL_CHAT_CONTEXT.scope( ApprovalChatContext { thread_id: "approval-deny-thread".to_string(), client_id: "approval-deny-client".to_string(), @@ -1387,8 +1433,9 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { ) .await }, - ) - .await + ), + ) + .await }); let deny_request_id = loop { @@ -1485,8 +1532,12 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { .await; assert!(decide_failure.get("error").is_some()); - let persist_failure = APPROVAL_CHAT_CONTEXT - .scope( + let persist_failure = turn_origin::with_origin( + AgentTurnOrigin::WebChat { + thread_id: "approval-persist-failure-thread".to_string(), + client_id: "approval-persist-failure-client".to_string(), + }, + APPROVAL_CHAT_CONTEXT.scope( ApprovalChatContext { thread_id: "approval-persist-failure-thread".to_string(), client_id: "approval-persist-failure-client".to_string(), @@ -1496,8 +1547,9 @@ async fn approval_rpc_decision_paths_persist_always_allow_and_recent_audit() { "tools.persistence_failure(action=coverage)", json!({ "action": "coverage" }), ), - ) - .await; + ), + ) + .await; match persist_failure.0 { GateOutcome::Deny { reason } => { assert!(reason.contains("Approval gate could not persist the request")); diff --git a/tests/worker_b_raw_coverage_e2e.rs b/tests/worker_b_raw_coverage_e2e.rs index ce8e7440f..0afaba629 100644 --- a/tests/worker_b_raw_coverage_e2e.rs +++ b/tests/worker_b_raw_coverage_e2e.rs @@ -21,6 +21,7 @@ use tempfile::{tempdir, TempDir}; use openhuman_core::core::auth::{init_rpc_token, CORE_TOKEN_ENV_VAR}; use openhuman_core::core::jsonrpc::build_core_http_router; +use openhuman_core::openhuman::agent::turn_origin::{self, AgentTurnOrigin}; use openhuman_core::openhuman::approval::gate::{ ApprovalChatContext, ApprovalGate, APPROVAL_CHAT_CONTEXT, }; @@ -585,8 +586,14 @@ async fn approval_gate_rpc_decision_resumes_parked_tool_and_records_execution() let gate_for_task = gate.clone(); let approval_task = tokio::spawn(async move { - APPROVAL_CHAT_CONTEXT - .scope( + // Scope a WebChat origin alongside the chat context — the gate now + // requires an origin label or it fails closed on `Unknown`. + turn_origin::with_origin( + AgentTurnOrigin::WebChat { + thread_id: "worker-b-thread".to_string(), + client_id: "worker-b-client".to_string(), + }, + APPROVAL_CHAT_CONTEXT.scope( ApprovalChatContext { thread_id: "worker-b-thread".to_string(), client_id: "worker-b-client".to_string(), @@ -600,8 +607,9 @@ async fn approval_gate_rpc_decision_resumes_parked_tool_and_records_execution() ) .await }, - ) - .await + ), + ) + .await }); let deadline = Instant::now() + Duration::from_secs(5);