feat(agent): persist provenance and gate sensitive egress

THREAT_MODEL.md

@@ -53,13 +53,35 @@ The agent may be running in a non-admin user's session, but tool dispatch first
 
 ## Prompt-Injection Hardening
 
-External content that reaches the LLM is treated as untrusted via `src/prompt_security.py`:
+Content provenance has two independent dimensions:
 
-- `untrusted_context_message(label, content)` wraps the content in a `user`-role message with a header block instructing the model not to follow instructions inside it. Content goes in as data, not as a system instruction.
+- **Integrity origin:** system, Odysseus-stored/user-editable, workspace/PR, or external.
+- **Sensitivity:** public, workspace, or private.
+
+External, workspace/PR, and Odysseus-stored content are all untrusted as instructions. Private is not an integrity claim; it means later egress can disclose user data even when the content itself is benign.
+
+Content that reaches the LLM is labelled via `src/prompt_security.py`:
+
+- `untrusted_context_message(label, content, origin=..., sensitivity=...)` wraps content in a `user`-role message with server-owned provenance metadata and a header block instructing the model not to follow instructions inside it. Content goes in as data, not as a system instruction.
 - `UNTRUSTED_CONTEXT_POLICY` is a system-prompt preamble that states the same policy at the top of every session where untrusted data may appear.
 
 **Untrusted surfaces that must go through this wrapper:** web search results, fetched URLs, emails (read), saved memories, skill text, notes, and any tool output sourced from outside the server. Injecting untrusted content directly into the system role is a security bug.
 
+The thread maintains monotonic `external`, `workspace`, `Odysseus`, and `private` observations in the database. Switching models, starting a later turn, teacher takeover, reloading the browser, or forking the conversation cannot clear them. Agent mode does not ambiently retrieve saved memories or personal-document RAG; private reads occur through explicit tool calls and their results are provenance-labelled before later actions. Directly opened documents, emails, and uploads are explicit selections and are labelled before the model sees them.
+
+### Agent Run Authority
+
+Model output requests an action; it does not authorize one. `src/tool_capabilities.py` classifies each built-in tool's effects and result integrity, while `src/agent_run_policy.py` combines those fixed classifications with the thread's server-owned security mode. The temporary blanket gate that armed after any untrusted tool result is disabled; mode-owned authority checks remain active.
+
+- **Sandbox (default):** process execution stays inside the workspace sandbox. Brokered public reads, private/workspace operations, and contained execution can continue without a new approval merely because earlier context was untrusted. Unknown tools, arbitrary network egress, external side effects, admin changes, and destructive actions still require an exact approval.
+- **Full access:** an admin or intentional single-user deployment may explicitly opt into direct execution with that user's normal OS permissions. This is never the default, and route, agent-loop, and dispatcher gates reject it for non-admin users.
+- **Ask (not exposed in the current selector):** uses the workspace sandbox and requires an exact approval for every risky action.
+
+Exact approvals are opaque, expiring, one-use server records bound to the owner, session, origin run, exact tool name and input, workspace, security mode, effect classification, and external-context state. The browser submits only the opaque approval ID and the user's approve/deny decision. The separate review step for saving a teacher-generated reusable skill remains active.
+Multiplexed tools such as `manage_memory`, `manage_documents`, and `manage_tasks` are classified from the sealed action discriminator. Known read actions get private-read authority, known writes get write authority, and malformed or unknown actions get the fail-high union.
+
+The dormant action-approval implementation uses opaque, expiring, one-use server records bound to the owner, session, origin run, exact tool name and input, workspace, security mode, effect classification, and complete provenance snapshot. It does not create automatic agent-action approvals while disabled. The separate review step for saving a teacher-generated reusable skill remains active.
+
 ## Security Headers
 
 `core/middleware.py:SecurityHeadersMiddleware` sets headers on every response:
@@ -72,7 +94,7 @@ External content that reaches the LLM is treated as untrusted via `src/prompt_se
 
 These are open, acknowledged, and contributor help is welcome:
 
-1. **No shell/filesystem sandbox.** The agent `bash` and `read_file`/`write_file` tools run as the app process user with no network egress filtering or filesystem confinement. A successful prompt-injection reaching a shell-enabled admin session can make outbound requests to internal services. See #1058 for the sandbox proposal.
+1. **Linux sandbox portability and explicit Full Access.** Agent `bash`, Python, tmux, and detached background commands default to a Bubblewrap profile with a private network namespace, cleared environment, private temp/home, one writable workspace, credential-path overlays, read-only `.git` metadata, and generous per-process rlimits. Internet-enabled process execution is limited to the trusted HTTP(S) egress broker; raw container networking is never shared with either mode. Odysseus performs the actual capability probes under the service user before process execution. A failed probe blocks only process tools and never downgrades automatically. An administrator may temporarily enable **Full Access** only after a warning plus typed confirmation. Full Access grants the process the Odysseus operating-system user's filesystem view while retaining the sandbox's private PID/network policy; an already-running Full Access process retains its launch-time authority until it exits or is killed. The process boundary mounts a fresh procfs scoped to its private PID namespace. Hard workspace-disk quotas and aggregate agent-pool/per-instance CPU, memory, and PID ceilings are not established in this slice; current limits are per process and the aggregate design is tracked separately.
 
 2. **SSRF via `/api/v1/chat` `base_url` parameter.** A chat-scoped API token can supply an arbitrary `base_url`; the server forwards the LLM request to that host without validating the scheme or address. PR #1039 fixes this.
 

core/database.py

@@ -4,9 +4,12 @@
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
-from urllib.parse import unquote, urlparse
+from src.sqlite_paths import (
+    normalize_sqlite_url as _normalize_sqlite_url_impl,
+    sqlite_db_path as _sqlite_db_path_impl,
+)
 from sqlalchemy import DDL, event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, inspect, text
-from sqlalchemy.engine import Engine, make_url
+from sqlalchemy.engine import Engine
 from sqlalchemy.types import TypeDecorator
 from sqlalchemy.ext.declarative import declarative_base, declared_attr
 from sqlalchemy.orm import relationship, sessionmaker, backref
@@ -45,28 +48,7 @@ def _default_database_url() -> str:
 
 
 def _normalize_sqlite_url(url: str) -> str:
-    """Resolve relative ordinary SQLite paths without rewriting URI filenames."""
-    try:
-        parsed = make_url(url)
-    except Exception:
-        return url
-
-    if parsed.get_backend_name() != "sqlite":
-        return url
-
-    db_path = parsed.database
-    if (
-        not db_path
-        or db_path == ":memory:"
-        or str(db_path).lower().startswith("file:")
-        or os.path.isabs(str(db_path))
-    ):
-        return url
-
-    absolute_path = (Path(get_app_root()) / str(db_path)).resolve().as_posix()
-    return parsed.set(database=absolute_path).render_as_string(
-        hide_password=False
-    )
+    return _normalize_sqlite_url_impl(url, app_root=get_app_root())
 
 
 # Get database URL from environment, default to SQLite in DATA_DIR
@@ -86,50 +68,7 @@ def _normalize_sqlite_url(url: str) -> str:
 
 
 def _sqlite_db_path(url) -> Optional[str]:
-    """Return the filesystem path for a file-backed SQLite URL.
-
-    SQLite query parameters such as ``mode=memory`` only affect filename
-    semantics when SQLAlchemy enables URI handling with ``uri=true``. Ordinary
-    file URLs must therefore remain file-backed even when they contain a query
-    parameter named ``mode``.
-
-    For SQLite ``file:`` URIs, an empty authority or ``localhost`` identifies a
-    local path. Other authorities are retained as UNC-style paths.
-    """
-    if url.get_backend_name() != "sqlite":
-        return None
-
-    db_path = url.database
-    if not db_path or db_path == ":memory:":
-        return None
-
-    db_path = str(db_path)
-    query = {
-        str(key).lower(): str(value).strip().lower()
-        for key, value in dict(getattr(url, "query", {}) or {}).items()
-    }
-    uri_enabled = query.get("uri") in {"1", "true", "yes", "on"}
-    is_file_uri = db_path.lower().startswith("file:")
-
-    if not uri_enabled or not is_file_uri:
-        return db_path
-
-    if (
-        db_path.lower().startswith("file::memory:")
-        or query.get("mode") == "memory"
-    ):
-        return None
-
-    parsed = urlparse(db_path)
-    fs_path = parsed.path or ""
-    if not fs_path or fs_path == ":memory:":
-        return None
-
-    authority = parsed.netloc
-    if authority and authority.lower() != "localhost":
-        fs_path = f"//{authority}{fs_path}"
-
-    return unquote(fs_path)
+    return _sqlite_db_path_impl(url)
 
 # Create session factory
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@@ -220,6 +159,11 @@ class Session(TimestampMixin, Base):
     total_input_tokens = Column(Integer, default=0)
     total_output_tokens = Column(Integer, default=0)
     mode = Column(String, nullable=True)  # 'agent', 'chat', or 'research'
+    security_mode = Column(String, nullable=False, default="sandbox")
+    agent_external_untrusted_seen = Column(Boolean, nullable=False, default=False)
+    agent_workspace_untrusted_seen = Column(Boolean, nullable=False, default=False)
+    agent_odysseus_untrusted_seen = Column(Boolean, nullable=False, default=False)
+    agent_private_data_seen = Column(Boolean, nullable=False, default=False)
     crew_member_id = Column(String, nullable=True)  # links to crew_members.id
 
     # Relationship to chat messages
@@ -249,6 +193,14 @@ def to_dict(self):
             'total_input_tokens': self.total_input_tokens or 0,
             'total_output_tokens': self.total_output_tokens or 0,
             'crew_member_id': self.crew_member_id,
+            'mode': self.mode,
+            'security_mode': self.security_mode or 'sandbox',
+            'agent_provenance': {
+                'external_untrusted_context_seen': bool(self.agent_external_untrusted_seen),
+                'workspace_untrusted_context_seen': bool(self.agent_workspace_untrusted_seen),
+                'odysseus_untrusted_context_seen': bool(self.agent_odysseus_untrusted_seen),
+                'private_data_context_seen': bool(self.agent_private_data_seen),
+            },
         }
 
 class ChatMessage(Base):
@@ -1259,6 +1211,60 @@ def _migrate_add_mode_column():
         except Exception:
             pass
 
+def _migrate_add_security_mode_column():
+    """Add the fail-safe agent run mode to existing session databases."""
+    try:
+        inspector = inspect(engine)
+        if "sessions" not in inspector.get_table_names():
+            return
+        columns = {column["name"] for column in inspector.get_columns("sessions")}
+        if "security_mode" not in columns:
+            with engine.begin() as conn:
+                conn.execute(text(
+                    "ALTER TABLE sessions ADD COLUMN security_mode VARCHAR "
+                    "NOT NULL DEFAULT 'sandbox'"
+                ))
+            logging.getLogger(__name__).info(
+                "Migrated: added 'security_mode' column to sessions"
+            )
+    except Exception as e:
+        logging.getLogger(__name__).warning(
+            f"Migration check for security_mode failed: {e}"
+        )
+
+
+def _migrate_add_agent_provenance_columns():
+    """Add monotonic per-thread provenance state to existing databases."""
+    columns_to_add = (
+        "agent_external_untrusted_seen",
+        "agent_workspace_untrusted_seen",
+        "agent_odysseus_untrusted_seen",
+        "agent_private_data_seen",
+    )
+    try:
+        inspector = inspect(engine)
+        if "sessions" not in inspector.get_table_names():
+            return
+        columns = {column["name"] for column in inspector.get_columns("sessions")}
+        changed = False
+        with engine.begin() as conn:
+            for column in columns_to_add:
+                if column not in columns:
+                    conn.execute(text(
+                        f"ALTER TABLE sessions ADD COLUMN {column} BOOLEAN "
+                        "NOT NULL DEFAULT FALSE"
+                    ))
+                    changed = True
+        if changed:
+            logging.getLogger(__name__).info(
+                "Migrated: added agent provenance columns to sessions"
+            )
+    except Exception as e:
+        logging.getLogger(__name__).warning(
+            f"Migration check for agent provenance failed: {e}"
+        )
+
+
 def _migrate_add_folder_column():
     """Add folder column to sessions table if it doesn't exist."""
     import sqlite3
@@ -2116,6 +2122,8 @@ def init_db():
     _migrate_add_folder_column()
     _migrate_add_token_columns()
     _migrate_add_mode_column()
+    _migrate_add_security_mode_column()
+    _migrate_add_agent_provenance_columns()
     _migrate_add_multiuser_owner_columns()
     _migrate_add_gallery_caption_column()
     _migrate_add_api_token_scopes_column()
@@ -2690,6 +2698,99 @@ def set_session_mode(session_id: str, mode: str) -> bool:
         logger.warning("Failed to persist mode %r for session %s", mode, session_id)
         return False
 
+def get_session_security_mode(session_id: str) -> str:
+    """Return the persisted agent authority mode, defaulting safely."""
+    try:
+        with get_db_session() as db:
+            value = db.query(Session.security_mode).filter(
+                Session.id == session_id
+            ).scalar()
+            return value if value in {"ask", "sandbox", "full_access"} else "sandbox"
+    except Exception:
+        logger.warning("Failed to read security mode for session %s", session_id)
+        return "sandbox"
+
+def set_session_security_mode(session_id: str, mode: str) -> bool:
+    """Persist a validated agent authority mode; invalid values fail closed."""
+    if mode not in {"ask", "sandbox", "full_access"}:
+        return False
+    try:
+        with get_db_session() as db:
+            db.query(Session).filter(Session.id == session_id).update(
+                {"security_mode": mode}
+            )
+        return True
+    except Exception:
+        logger.warning(
+            "Failed to persist security mode %r for session %s",
+            mode,
+            session_id,
+        )
+        return False
+
+def get_session_agent_provenance(session_id: str) -> dict:
+    """Return server-owned monotonic provenance state for one thread."""
+    empty = {
+        "external_untrusted_context_seen": False,
+        "workspace_untrusted_context_seen": False,
+        "odysseus_untrusted_context_seen": False,
+        "private_data_context_seen": False,
+    }
+    try:
+        with get_db_session() as db:
+            row = db.query(
+                Session.agent_external_untrusted_seen,
+                Session.agent_workspace_untrusted_seen,
+                Session.agent_odysseus_untrusted_seen,
+                Session.agent_private_data_seen,
+            ).filter(Session.id == session_id).first()
+            if row is None:
+                return empty
+            return {
+                "external_untrusted_context_seen": bool(row[0]),
+                "workspace_untrusted_context_seen": bool(row[1]),
+                "odysseus_untrusted_context_seen": bool(row[2]),
+                "private_data_context_seen": bool(row[3]),
+            }
+    except Exception:
+        logger.warning("Failed to read agent provenance for session %s", session_id)
+        return empty
+
+def merge_session_agent_provenance(session_id: str, state) -> bool:
+    """Persist only newly observed provenance bits; never clear thread state."""
+    if not session_id:
+        return False
+    try:
+        from src.provenance import ConversationProvenance
+        provenance = (
+            state
+            if isinstance(state, ConversationProvenance)
+            else ConversationProvenance.from_mapping(state)
+        )
+        values = {}
+        if provenance.external_untrusted_context_seen:
+            values["agent_external_untrusted_seen"] = True
+        if provenance.workspace_untrusted_context_seen:
+            values["agent_workspace_untrusted_seen"] = True
+        if provenance.odysseus_untrusted_context_seen:
+            values["agent_odysseus_untrusted_seen"] = True
+        if provenance.private_data_context_seen:
+            values["agent_private_data_seen"] = True
+        if not values:
+            return True
+        with get_db_session() as db:
+            updated = db.query(Session).filter(Session.id == session_id).update(
+                values,
+                synchronize_session=False,
+            )
+        return bool(updated)
+    except Exception:
+        logger.warning(
+            "Failed to persist agent provenance for session %s",
+            session_id,
+        )
+        return False
+
 def get_session_by_id(session_id: str):
     """Get a session by ID"""
     with get_db_session() as db:

core/models.py

@@ -74,13 +74,22 @@ class Session:
     owner: Optional[str] = None
     is_important: bool = False
     message_count: int = 0
+    security_mode: str = "sandbox"
+    agent_provenance: Optional[Dict[str, bool]] = None
 
     def __post_init__(self):
         if self.headers is None:
             self.headers = {}
         # Ensure each session gets its OWN list (not the shared dataclass default)
         if self.history is None:
             self.history = []
+        if self.agent_provenance is None:
+            self.agent_provenance = {
+                "external_untrusted_context_seen": False,
+                "workspace_untrusted_context_seen": False,
+                "odysseus_untrusted_context_seen": False,
+                "private_data_context_seen": False,
+            }
 
     @property
     def _history(self) -> List[ChatMessage]: