diff --git a/ROADMAP.md b/ROADMAP.md index 7c59c1f6a6..d29ac5c752 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -12,7 +12,6 @@ the codebase, you are probably right to stay away. and WSL all need coverage. - Integration audit: do integrations even work? Confirm what works, what needs setup docs, and what should be removed or hidden. -- Self-host troubleshooting cookbook. Document the weird 30-second fixes that otherwise become 30-minute searches: Dovecot cleartext auth for local stacks, ntfy Android Instant Delivery for non-ntfy.sh servers, clipboard limits on plain-HTTP Tailscale URLs, Radicale collection URLs, and similar traps. - Cookbook reliability on other computers. This is probably the area most likely to need work across different machines, GPUs, drivers, shells, and Python environments. - Cookbook SGLang support across platforms. Make sure SGLang setup/serve works predictably on Linux, Windows/WSL, macOS where possible, Docker, and common diff --git a/app.py b/app.py index a89f80143c..38c848e0bd 100644 --- a/app.py +++ b/app.py @@ -655,7 +655,12 @@ async def _stop_background(): # Sessions from routes.session_routes import setup_session_routes session_config = {"REQUEST_TIMEOUT": REQUEST_TIMEOUT, "OPENAI_API_KEY": OPENAI_API_KEY, "SESSIONS_FILE": SESSIONS_FILE} -app.include_router(setup_session_routes(session_manager, session_config, webhook_manager=webhook_manager)) +app.include_router(setup_session_routes( + session_manager, + session_config, + webhook_manager=webhook_manager, + upload_handler=upload_handler, +)) # Admin Danger Zone wipes (Settings → System → Danger Zone) from routes.admin_wipe_routes import setup_admin_wipe_routes @@ -684,7 +689,7 @@ async def _stop_background(): # History from routes.history.history_routes import setup_history_routes -app.include_router(setup_history_routes(session_manager)) +app.include_router(setup_history_routes(session_manager, upload_handler=upload_handler)) # Search from routes.search_routes import setup_search_routes @@ -763,7 +768,7 @@ async def _stop_background(): # Calendar (CalDAV) from routes.calendar_routes import setup_calendar_routes -calendar_router = setup_calendar_routes() +calendar_router = setup_calendar_routes(upload_handler=upload_handler) app.include_router(calendar_router) # Shell (user-facing command execution) @@ -826,7 +831,7 @@ async def _stop_background(): # Notes (Google Keep-style notes/todos) from routes.note_routes import setup_note_routes -app.include_router(setup_note_routes(task_scheduler)) +app.include_router(setup_note_routes(task_scheduler, upload_handler=upload_handler)) # Email from routes.email_routes import setup_email_routes @@ -1043,7 +1048,7 @@ async def _startup_mcp_connections(): except BaseException as e: logger.warning(f"Built-in MCP registration failed (non-critical): {type(e).__name__}: {e}") try: - await asyncio.wait_for(mcp_manager.connect_all_enabled(), timeout=20) + await mcp_manager.connect_all_enabled() except asyncio.TimeoutError: logger.warning("User MCP startup timed out (non-critical)") except BaseException as e: diff --git a/core/database.py b/core/database.py index ade995871a..a9ad90b8b7 100644 --- a/core/database.py +++ b/core/database.py @@ -3,13 +3,16 @@ import sqlite3 from datetime import datetime, timezone from pathlib import Path +from typing import Optional +from urllib.parse import unquote, urlparse from sqlalchemy import event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, text -from sqlalchemy.engine import Engine +from sqlalchemy.engine import Engine, make_url from sqlalchemy.types import TypeDecorator from sqlalchemy.ext.declarative import declarative_base, declared_attr from sqlalchemy.orm import relationship, sessionmaker, backref from src.runtime_paths import get_app_root +from core.platform_compat import safe_chmod, IS_WINDOWS logger = logging.getLogger(__name__) @@ -42,12 +45,28 @@ def _default_database_url() -> str: def _normalize_sqlite_url(url: str) -> str: - if not url.startswith("sqlite:///"): + """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 = url.replace("sqlite:///", "", 1) - if db_path == ":memory:" or os.path.isabs(db_path): + + 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 - return f"sqlite:///{(Path(get_app_root()) / db_path).resolve().as_posix()}" + + absolute_path = (Path(get_app_root()) / str(db_path)).resolve().as_posix() + return parsed.set(database=absolute_path).render_as_string( + hide_password=False + ) # Get database URL from environment, default to SQLite in DATA_DIR @@ -59,6 +78,59 @@ def _normalize_sqlite_url(url: str) -> str: connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {} ) + +# Sidecar files SQLite can create next to the main DB. -journal is the default +# rollback journal; -wal/-shm appear once WAL is enabled. Each can hold copies of +# secret-bearing pages, so they get the same 0o600 lockdown as the DB itself. +_SQLITE_SIDECARS = ("-journal", "-wal", "-shm") + + +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) + # Create session factory SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) @@ -1819,6 +1891,41 @@ def init_db(): """ _migrate_model_endpoints() Base.metadata.create_all(bind=engine) + # Lock the DB file (and any SQLite sidecars) to 0o600 — it holds bearer-token + # + bcrypt hashes and encrypted provider keys. POSIX only; safe_chmod no-ops + # on Windows (ACL-restricted profile dir) and the path helper returns None for + # Postgres / in-memory. Must stay AFTER create_all: the file is born here at + # the umask default, and nothing below resets the mode. The path comes from + # engine.url (SQLAlchemy's parsed URL), so a driver-qualified or query-tagged + # DATABASE_URL still resolves to the real file instead of slipping through. + db_path = _sqlite_db_path(engine.url) + if db_path is not None: + # Fail closed-loud on the main file: this is the only access control on + # it, so if the chmod genuinely fails (read-only FS, foreign owner) an + # operator should hear about it. safe_chmod also returns False as a + # Windows no-op, so guard on IS_WINDOWS to avoid a spurious warning there. + if not safe_chmod(db_path, 0o600) and not IS_WINDOWS: + logger.warning( + "Could not restrict %s to 0o600; it holds secrets and may be " + "world-readable. Check filesystem permissions and ownership.", + db_path, + ) + # Re-lock any sidecars present at startup. New ones inherit the main + # file's mode (now 0o600, since we set it first), and they're usually + # absent here, but a stale -wal/-shm/-journal left by an older 0o644 + # install could still expose secret pages. Absent sidecars are the + # normal case, not an error — only a failed chmod warrants a warning. + for suffix in _SQLITE_SIDECARS: + sidecar = db_path + suffix + if ( + os.path.exists(sidecar) + and not safe_chmod(sidecar, 0o600) + and not IS_WINDOWS + ): + logger.warning( + "Could not restrict %s to 0o600; it may expose DB pages.", + sidecar, + ) _migrate_add_hidden_models_column() _migrate_add_cached_models_column() _migrate_add_pinned_models_column() @@ -1904,6 +2011,20 @@ def _migrate_chat_messages_fts(): conn = None try: conn = sqlite3.connect(db_path) + fts_content_expr_new = ( + "CASE WHEN instr(COALESCE(new.content, ''), ';base64,') > 0 " + "OR instr(COALESCE(new.content, ''), 'data:image/') > 0 " + "OR instr(COALESCE(new.content, ''), 'data:audio/') > 0 " + "THEN '[inline media omitted from search index]' " + "ELSE COALESCE(new.content, '') END" + ) + fts_content_expr_cm = ( + "CASE WHEN instr(COALESCE(cm.content, ''), ';base64,') > 0 " + "OR instr(COALESCE(cm.content, ''), 'data:image/') > 0 " + "OR instr(COALESCE(cm.content, ''), 'data:audio/') > 0 " + "THEN '[inline media omitted from search index]' " + "ELSE COALESCE(cm.content, '') END" + ) try: conn.execute("CREATE VIRTUAL TABLE IF NOT EXISTS temp._odysseus_fts5_probe USING fts5(content)") conn.execute("DROP TABLE IF EXISTS temp._odysseus_fts5_probe") @@ -1912,7 +2033,7 @@ def _migrate_chat_messages_fts(): return conn.executescript( - """ + f""" CREATE VIRTUAL TABLE IF NOT EXISTS chat_messages_fts USING fts5( content, message_id UNINDEXED, @@ -1920,10 +2041,14 @@ def _migrate_chat_messages_fts(): role UNINDEXED ); + DROP TRIGGER IF EXISTS chat_messages_fts_ai; + DROP TRIGGER IF EXISTS chat_messages_fts_ad; + DROP TRIGGER IF EXISTS chat_messages_fts_au; + CREATE TRIGGER IF NOT EXISTS chat_messages_fts_ai AFTER INSERT ON chat_messages BEGIN INSERT INTO chat_messages_fts(content, message_id, session_id, role) - VALUES (COALESCE(new.content, ''), new.id, new.session_id, new.role); + VALUES ({fts_content_expr_new}, new.id, new.session_id, new.role); END; CREATE TRIGGER IF NOT EXISTS chat_messages_fts_ad @@ -1935,14 +2060,14 @@ def _migrate_chat_messages_fts(): AFTER UPDATE ON chat_messages BEGIN DELETE FROM chat_messages_fts WHERE message_id = old.id; INSERT INTO chat_messages_fts(content, message_id, session_id, role) - VALUES (COALESCE(new.content, ''), new.id, new.session_id, new.role); + VALUES ({fts_content_expr_new}, new.id, new.session_id, new.role); END; """ ) conn.execute( - """ + f""" INSERT INTO chat_messages_fts(content, message_id, session_id, role) - SELECT COALESCE(cm.content, ''), cm.id, cm.session_id, cm.role + SELECT {fts_content_expr_cm}, cm.id, cm.session_id, cm.role FROM chat_messages cm WHERE NOT EXISTS ( SELECT 1 FROM chat_messages_fts fts @@ -1950,6 +2075,7 @@ def _migrate_chat_messages_fts(): ) """ ) + _scrub_legacy_chat_message_fts_media(conn) conn.commit() except Exception as e: logging.getLogger(__name__).warning(f"chat_messages FTS migration failed: {e}") @@ -1960,6 +2086,37 @@ def _migrate_chat_messages_fts(): pass +def _scrub_legacy_chat_message_fts_media(conn) -> None: + """Replace already-indexed inline media rows with searchable text only.""" + try: + from src.attachment_refs import search_index_text + except Exception as e: + logging.getLogger(__name__).warning(f"chat_messages FTS media scrub skipped: {e}") + return + + try: + rows = conn.execute( + """ + SELECT id, session_id, role, content + FROM chat_messages + WHERE instr(COALESCE(content, ''), ';base64,') > 0 + OR instr(COALESCE(content, ''), 'data:image/') > 0 + OR instr(COALESCE(content, ''), 'data:audio/') > 0 + """ + ).fetchall() + for message_id, session_id, role, content in rows: + conn.execute("DELETE FROM chat_messages_fts WHERE message_id = ?", (message_id,)) + conn.execute( + """ + INSERT INTO chat_messages_fts(content, message_id, session_id, role) + VALUES (?, ?, ?, ?) + """, + (search_index_text(content), message_id, session_id, role), + ) + except Exception as e: + logging.getLogger(__name__).warning(f"chat_messages FTS media scrub failed: {e}") + + def _migrate_add_email_smtp_security(): """Add explicit SMTP security mode for Proton Bridge/custom local SMTP.""" import sqlite3 diff --git a/core/session_manager.py b/core/session_manager.py index 491fbc0782..f5024d212c 100644 --- a/core/session_manager.py +++ b/core/session_manager.py @@ -16,6 +16,8 @@ from .database import Session as DbSession, ChatMessage as DbChatMessage, Document as DbDocument, SessionLocal, utcnow_naive from .models import Session, ChatMessage +from src.attachment_refs import persistable_message_content +from src.upload_handler import reserve_message_upload_references # Re-export singleton accessors from models for convenience from .models import set_session_manager_instance, get_session_manager_instance @@ -72,6 +74,7 @@ class SessionManager: def __init__(self, sessions_file: str = None): # sessions_file kept for backward compat, not used self.sessions: Dict[str, Session] = {} + self.upload_handler = None self.load_sessions() # ------------------------------------------------------------------ @@ -230,17 +233,26 @@ def _persist_message(self, session_id: str, message: ChatMessage): logger.warning("Dropping message for deleted session %s", session_id) return + missing_upload_id = reserve_message_upload_references( + getattr(self, "upload_handler", None), + getattr(db_session, "owner", None), + message.content, + message.metadata, + ) + if missing_upload_id: + raise ValueError( + f"Referenced upload is no longer available: {missing_upload_id}" + ) + msg_id = str(uuid.uuid4()) msg_time = datetime.utcnow() if message.metadata is None: message.metadata = {} message.metadata.setdefault('timestamp', _message_timestamp_iso(msg_time)) - # Multimodal content (image/audio attachments) is a list — serialize - # to JSON so the Text column can store it. On reload, _db_to_session - # detects the JSON-array prefix and parses it back. - _content = message.content - if isinstance(_content, list): - _content = json.dumps(_content) + # Multimodal content may contain provider data URLs for the live + # model call. Persist only readable text plus attachment references + # so chat_messages/FTS do not duplicate upload bytes. + _content = persistable_message_content(message.content, message.metadata) db_message = DbChatMessage( id=msg_id, session_id=session_id, @@ -322,6 +334,28 @@ def replace_messages(self, session_id: str, messages: list) -> bool: session = self.get_session(session_id) db = SessionLocal() try: + db_session = db.query(DbSession).filter(DbSession.id == session_id).first() + if db_session is None: + logger.warning("Cannot replace history for missing session %s", session_id) + return False + + # Reserve every incoming attachment before removing any durable + # message row. reserve_upload() shares the upload lifecycle lock + # with cleanup, so an upload cannot be deleted between this + # ownership check/access touch and the replacement transaction. + # A failed reservation must leave the existing transcript intact. + for message in messages: + missing_upload_id = reserve_message_upload_references( + getattr(self, "upload_handler", None), + getattr(db_session, "owner", None), + message.content, + message.metadata, + ) + if missing_upload_id: + raise ValueError( + f"Referenced upload is no longer available: {missing_upload_id}" + ) + db.query(DbChatMessage).filter(DbChatMessage.session_id == session_id).delete() now = datetime.now(timezone.utc) for i, message in enumerate(messages): @@ -330,15 +364,9 @@ def replace_messages(self, session_id: str, messages: list) -> bool: id=msg_id, session_id=session_id, role=message.role, - # Multimodal content (image/audio attachments) is a list; - # serialize to JSON so the Text column round-trips via - # _parse_msg_content. Storing the raw list let SQLAlchemy - # bind its single-quoted repr, which _parse_msg_content - # cannot parse (it looks for double-quoted "type"), so the - # attachment was destroyed on reload. Mirrors _persist_message. - content=(json.dumps(message.content) - if isinstance(message.content, list) - else message.content), + # Mirrors _persist_message: keep raw media bytes out of the + # persisted transcript and search index. + content=persistable_message_content(message.content, message.metadata), meta_data=json.dumps(message.metadata) if message.metadata else None, timestamp=now + timedelta(microseconds=i), ) @@ -347,12 +375,10 @@ def replace_messages(self, session_id: str, messages: list) -> bool: message.metadata = {} message.metadata["_db_id"] = msg_id - db_session = db.query(DbSession).filter(DbSession.id == session_id).first() - if db_session: - db_session.message_count = len(messages) - db_session.updated_at = now - db_session.last_accessed = now - db_session.last_message_at = now + db_session.message_count = len(messages) + db_session.updated_at = now + db_session.last_accessed = now + db_session.last_message_at = now db.commit() session.history = list(messages) diff --git a/docs/attachments.md b/docs/attachments.md new file mode 100644 index 0000000000..93f9e0ffe1 --- /dev/null +++ b/docs/attachments.md @@ -0,0 +1,85 @@ +# Attachment References and Upload Storage + +Odysseus stores uploaded bytes once under the configured upload directory and +passes stable references through chat history, tools, and future artifact work. +The goal is to avoid duplicating large inline media payloads in +`chat_messages.content` or the SQLite FTS index. + +## Reference Shape + +Attachment references use this minimum shape: + +```json +{ + "type": "attachment_ref", + "attachment_id": "32hex-or-32hex.ext", + "name": "original-filename.png", + "mime": "image/png", + "size": 12345, + "checksum_sha256": "hex-digest", + "created_at": "2026-07-09T12:00:00" +} +``` + +Optional fields such as `width`, `height`, `vision`, `vision_model`, and +`gallery_id` may be present when the uploader or preprocessing path knows them. + +## Persistence + +The live model call may still receive provider-specific multimodal blocks for +the current turn. Persistence is different: + +- `chat_messages.content` stores readable text plus compact attachment reference + lines, never raw `data:*;base64,...` upload bytes. +- `chat_messages.metadata.attachments` stores structured attachment reference + metadata for UI reloads and future processing. +- The SQLite FTS migration recreates chat-message FTS triggers so new rows do + not index inline media payloads, and it scrubs legacy rows that were already + indexed with data URLs. + +## Tool Access + +Agent/tool context receives upload entries as `attachment_ref` manifests with an +`odysseus://attachment/` URI and `read_policy: "owner_checked_upload"`. + +For compatibility with existing built-in tools, a local `path` may be included +only after all of these checks pass: + +- the upload ID resolves through `UploadHandler.resolve_upload`; +- the requested owner is allowed to read the upload; +- the file remains inside the configured upload directory; +- the file path is inside the tool-readable roots. + +External MCP/custom tools should treat the URI and attachment ID as the stable +contract and request bytes through an owner-checked server path, not by assuming +host filesystem layout. + +## Retention and Deletion + +Current retention behavior is conservative: + +- uploads are indexed in `uploads.json` with owner, checksum, MIME type, size, + and creation time; +- admin cleanup first scans persisted chat metadata/content, document versions, + PDF source markers, gallery hashes, notes, and calendar records for live + references; +- cleanup fails closed if that reference scan cannot complete, and the lower-level + cleanup API removes nothing unless it receives a complete reference snapshot; +- expired, unreferenced uploads are removed during the completed scan, while + attachment-bearing writers must first take an owner-checked reservation that + serializes with deletion and refreshes the upload's access timestamp; +- deliberate removal atomically drops matching `uploads.json` rows before deleting + the bytes and restores those rows if filesystem removal fails; +- deleting a chat removes the chat rows but does not immediately delete shared + upload bytes, because the same upload may also be referenced by gallery items, + documents, duplicate-upload rows, or future artifact records. + +There is no distinct artifact table in the current schema. Artifact-like upload +references persisted in chat or document text are covered by the canonical +attachment-ID scan; any future artifact store must be added to reference discovery +before cleanup is allowed to consider its uploads unreferenced. + +Cleanup and write reservations share the upload-index lock. This closes the +scan/write/delete race in the documented single-worker deployment; a future +multi-process deployment must add an inter-process lock or move lifecycle state +into the database before enabling destructive cleanup in more than one worker. diff --git a/routes/calendar_routes.py b/routes/calendar_routes.py index 81dd48a5b2..6e0ee124c3 100644 --- a/routes/calendar_routes.py +++ b/routes/calendar_routes.py @@ -13,8 +13,9 @@ from dateutil.rrule import rrulestr from core.database import SessionLocal, CalendarCal, CalendarDeletedEvent, CalendarEvent -from src.auth_helpers import require_user +from src.auth_helpers import effective_user, require_user from src.upload_limits import read_upload_limited, ICS_MAX_BYTES +from src.upload_handler import reserve_upload_references logger = logging.getLogger(__name__) @@ -697,9 +698,18 @@ def _expand_rrule( # ── Routes ── -def setup_calendar_routes() -> APIRouter: +def setup_calendar_routes(upload_handler=None) -> APIRouter: router = APIRouter(prefix="/api/calendar", tags=["calendar"]) + def _reserve_calendar_uploads(request: Request, *values) -> None: + missing_id = reserve_upload_references( + upload_handler, + effective_user(request), + *values, + ) + if missing_id: + raise HTTPException(409, f"Referenced upload is no longer available: {missing_id}") + # ── CalDAV multi-account helpers ───────────────────────────────────────── def _get_caldav_accounts(owner: str) -> list: @@ -913,7 +923,24 @@ async def test_connection(request: Request): '' ) try: - async with httpx.AsyncClient(timeout=8.0, follow_redirects=False, trust_env=False) as cx: + # Build an SSL context that trusts the operator's custom CA bundle + # (SSL_CERT_FILE / REQUESTS_CA_BUNDLE) so self-signed CalDAV servers + # pass the pre-flight the same way they pass the real sync. + # trust_env=False is kept to block proxy/auth env leakage; the CA + # bundle is loaded explicitly instead. + import ssl as _ssl + _ssl_ctx = _ssl.create_default_context() + # Disable VERIFY_X509_STRICT so certs without a keyUsage extension + # (common in self-signed setups) are accepted, matching the + # requests/urllib3 behavior used by the CalDAV sync path. + _ssl_ctx.verify_flags &= ~_ssl.VERIFY_X509_STRICT + _ca_bundle = _os.environ.get("SSL_CERT_FILE") or _os.environ.get("REQUESTS_CA_BUNDLE") + if _ca_bundle: + if _os.path.isfile(_ca_bundle): + _ssl_ctx.load_verify_locations(_ca_bundle) + else: + logger.warning("CalDAV test: CA bundle %s not found, using system CAs", _ca_bundle) + async with httpx.AsyncClient(timeout=8.0, follow_redirects=False, trust_env=False, verify=_ssl_ctx) as cx: r = await cx.request( "PROPFIND", url, auth=(user, pw), @@ -1070,6 +1097,7 @@ async def list_events(request: Request, start: str, end: str, calendar: str = "" @router.post("/events") async def create_event(request: Request, data: EventCreate): owner = _require_user(request) + _reserve_calendar_uploads(request, data.color, data.description, data.location) db = SessionLocal() try: cal = None @@ -1131,6 +1159,7 @@ async def create_event(request: Request, data: EventCreate): @router.put("/events/{uid}") async def update_event(request: Request, uid: str, data: EventUpdate): owner = _require_user(request) + _reserve_calendar_uploads(request, data.color, data.description, data.location) try: base_uid = _resolve_base_uid(uid) except ValueError as e: @@ -1224,6 +1253,7 @@ async def delete_event(request: Request, uid: str, scope: str = "series"): @router.post("/calendars") async def create_calendar(request: Request, name: str = "Imported", color: str = "#5b8abf"): owner = _require_user(request) + _reserve_calendar_uploads(request, color) db = SessionLocal() try: cal = CalendarCal( @@ -1246,6 +1276,7 @@ async def create_calendar(request: Request, name: str = "Imported", color: str = @router.put("/calendars/{cal_id}") async def update_calendar(request: Request, cal_id: str, name: str = None, color: str = None): owner = _require_user(request) + _reserve_calendar_uploads(request, color) db = SessionLocal() try: cal = _get_or_404_calendar(db, cal_id, owner) diff --git a/routes/chat_helpers.py b/routes/chat_helpers.py index cdd204ec26..63c8abc8f8 100644 --- a/routes/chat_helpers.py +++ b/routes/chat_helpers.py @@ -17,6 +17,7 @@ from src.model_context import estimate_tokens from src.auth_helpers import effective_user from src.prompt_security import untrusted_context_message +from src.attachment_refs import attachment_ref from routes.prefs_routes import _load_for_user as load_prefs_for_user from fastapi import HTTPException @@ -418,13 +419,16 @@ def _read_file_can_open(path: str) -> bool: except Exception: path = None - manifest.append({ - "id": info.get("id") or str(att_id), - "name": info.get("name") or info.get("original_name") or str(att_id), - "mime": info.get("mime", ""), - "size": info.get("size", 0), + ref = attachment_ref({**info, "id": info.get("id") or str(att_id)}) + ref.update({ + "id": ref["attachment_id"], + "uri": f"odysseus://attachment/{ref['attachment_id']}", + "read_policy": "owner_checked_upload", + # Transitional compatibility: existing built-in tools can still use + # this path, but only after owner, upload-root, and tool-root checks. "path": path, }) + manifest.append(ref) return manifest diff --git a/routes/chat_routes.py b/routes/chat_routes.py index 705fa4ba96..b8d9934b4c 100644 --- a/routes/chat_routes.py +++ b/routes/chat_routes.py @@ -42,7 +42,12 @@ _enforce_chat_privileges, ) from src.action_intents import ToolIntent, classify_tool_intent as _classify_tool_intent -from src.tool_policy import build_effective_tool_policy +from src.tool_policy import ( + WEB_TOOL_NAMES, + build_effective_tool_policy, + is_web_search_explicitly_denied, + web_search_enabled_for_turn, +) logger = logging.getLogger(__name__) @@ -583,10 +588,7 @@ async def chat_stream(request: Request) -> StreamingResponse: # below). Skill extraction should only learn from real agent sessions, # not chats we quietly promoted for a notes/calendar intent. user_requested_agent = (chat_mode == "agent") - _search_enabled = ( - str(allow_web_search).lower() == "true" - or str(use_web).lower() == "true" - ) + _search_enabled = web_search_enabled_for_turn(allow_web_search, use_web) # Intent auto-escalation: if the user is clearly asking the assistant # to create a todo, reminder, or calendar event, promote chat → agent # for this turn so the LLM has access to manage_notes / manage_calendar. @@ -870,23 +872,20 @@ async def chat_stream(request: Request) -> StreamingResponse: # Build disabled-tools set from frontend toggles + user privileges disabled_tools = set() - # Only disable bash/web_search when the caller *explicitly* set them - # to a falsy value. When unset (None), defer to per-user privilege - # checks below — this lets admins with can_use_bash=True use bash - # by default without having to send allow_bash in every request. + # Only disable bash when the caller *explicitly* set it to a falsy + # value. When unset (None), defer to per-user privilege checks below. + # Web search is per-turn opt-in: either the chat pre-search setting + # (`use_web=true`) or agent web toggle (`allow_web_search=true`) must + # explicitly enable it. if allow_bash is not None and str(allow_bash).lower() != "true": disabled_tools.add("bash") - if ( - allow_web_search is not None - and str(allow_web_search).lower() != "true" - ): - disabled_tools.add("web_search") - disabled_tools.add("web_fetch") + _explicit_web_intent = bool(_tool_intent and _tool_intent.category == "web") + if is_web_search_explicitly_denied(allow_web_search) or not _search_enabled: + disabled_tools.update(WEB_TOOL_NAMES) if _explicit_web_intent: # A direct lookup/search request should not drift into personal - # tools or shell fallbacks. We still keep web_search/web_fetch - # available even when the frontend toggle is stale/falsy because - # the user's words are the stronger signal. + # tools or shell fallbacks. It can only use web_search/web_fetch + # when the request's explicit web setting enabled them. disabled_tools.update({ "bash", "python", "search_chats", "manage_skills", "manage_memory", @@ -896,11 +895,12 @@ async def chat_stream(request: Request) -> StreamingResponse: "manage_notes", "manage_calendar", "manage_tasks", "api_call", "builtin_browser", }) - disabled_tools.discard("web_search") - disabled_tools.discard("web_fetch") + if _search_enabled: + disabled_tools.difference_update(WEB_TOOL_NAMES) + else: + disabled_tools.update(WEB_TOOL_NAMES) elif _search_enabled: - disabled_tools.discard("web_search") - disabled_tools.discard("web_fetch") + disabled_tools.difference_update(WEB_TOOL_NAMES) # Nobody/incognito mode: deny tools that would expose the user's # persistent memory, past chats, or other identity-linked data. @@ -951,14 +951,7 @@ async def chat_stream(request: Request) -> StreamingResponse: from src.settings import get_setting _global_disabled = get_setting("disabled_tools", []) if _global_disabled and isinstance(_global_disabled, list): - explicit_web_allowed = ( - _explicit_web_intent - or (allow_web_search is not None and str(allow_web_search).lower() == "true") - ) - if explicit_web_allowed: - disabled_tools.update(t for t in _global_disabled if t not in {"web_search", "web_fetch"}) - else: - disabled_tools.update(_global_disabled) + disabled_tools.update(_global_disabled) # Light auto-escalation: the user is in chat mode and just expressed a # notes/calendar/email intent. Grant the relevant managers but withhold @@ -1410,10 +1403,8 @@ def _on_research_done(_sid, _result, _sources, _findings): _max_rounds = max(1, min(_max_rounds, 200)) _forced_tools = None - if _explicit_web_intent: - _forced_tools = {"web_search", "web_fetch"} - elif _search_enabled: - _forced_tools = {"web_search", "web_fetch"} + if _search_enabled: + _forced_tools = set(WEB_TOOL_NAMES) async for chunk in stream_agent_loop( sess.endpoint_url, @@ -1459,7 +1450,9 @@ def _on_research_done(_sid, _result, _sources, _findings): "tool_start", "tool_output", "agent_step", "doc_stream_open", "doc_stream_delta", "doc_update", "doc_suggestions", "ui_control", - "rounds_exhausted", + "rounds_exhausted", "budget_exceeded", + "loop_breaker_triggered", + "intent_nudge_exhausted", "ask_user", "plan_update", ): diff --git a/routes/document_routes.py b/routes/document_routes.py index e1b395e724..dae8b09fa2 100644 --- a/routes/document_routes.py +++ b/routes/document_routes.py @@ -12,6 +12,7 @@ from core.database import Session as DbSession from src.auth_helpers import get_current_user, _auth_disabled from src.constants import MAIL_ATTACHMENTS_DIR +from src.upload_handler import reserve_upload_references logger = logging.getLogger(__name__) @@ -78,6 +79,14 @@ def _email_source_key(content: str) -> tuple[str, str]: def setup_document_routes(session_manager, upload_handler=None) -> APIRouter: router = APIRouter(tags=["documents"]) + def _reserve_document_uploads(user: Optional[str], content: str) -> None: + missing_id = reserve_upload_references(upload_handler, user, content) + if missing_id: + raise HTTPException( + 409, + f"Referenced upload is no longer available: {missing_id}", + ) + def _locate_current_user_upload(request: Request, upload_id: str, user: Optional[str]): if upload_handler is None: return None @@ -124,6 +133,7 @@ async def create_document(request: Request, req: DocumentCreate) -> Dict[str, An if _looks_like_email_document(req.content, req.title): language = "email" + _reserve_document_uploads(user, req.content) _assert_pdf_marker_upload_owned(request, req.content, user, upload_handler) # Reply drafts are keyed to the source email. If a UI/tool path tries @@ -636,6 +646,7 @@ async def update_document(request: Request, doc_id: str, req: DocumentUpdate) -> if doc.current_content == incoming_content and not req.force_version: return _doc_to_dict(doc) + _reserve_document_uploads(user, incoming_content) _assert_pdf_marker_upload_owned(request, incoming_content, user, upload_handler) # Check if we can coalesce with the latest version diff --git a/routes/email_pollers.py b/routes/email_pollers.py index 13f846a710..8408103daa 100644 --- a/routes/email_pollers.py +++ b/routes/email_pollers.py @@ -1068,6 +1068,28 @@ def _scheduled_poll_once() -> dict: for r in rows: sid = r[0] try: + # Atomically claim this row before doing any work. Two + # pollers can race here (the in-process asyncio task and an + # externally cron-driven `odysseus-mail poll-scheduled`, or + # an admin running the CLI manually alongside the in-process + # one despite the ODYSSEUS_INPROCESS_POLLERS=0 guidance) - + # both can SELECT the same 'pending' row before either has + # updated its status. The UPDATE...WHERE status='pending' is + # the atomicity boundary: only the poller whose UPDATE + # actually changes a row (rowcount == 1) proceeds to send; + # a loser sees rowcount == 0 and skips it instead of sending + # a duplicate. + claim_conn = sqlite3.connect(SCHEDULED_DB) + claim_cur = claim_conn.execute( + "UPDATE scheduled_emails SET status='sending' WHERE id=? AND status='pending'", + (sid,), + ) + claim_conn.commit() + claimed = claim_cur.rowcount == 1 + claim_conn.close() + if not claimed: + continue + attachments = json.loads(r[8] or "[]") row_account_id = r[9] if len(r) > 9 else None odysseus_kind = r[10] if len(r) > 10 else "scheduled" diff --git a/routes/email_routes.py b/routes/email_routes.py index 9bbf7eda0b..fba679cc18 100644 --- a/routes/email_routes.py +++ b/routes/email_routes.py @@ -923,29 +923,32 @@ def _resolve_send_config(account_id: str | None = None, owner: str = "") -> dict def _store_email_flag(conn, uid: str, flag: str, add: bool = True) -> bool: + # imaplib's plain store() takes a message SEQUENCE NUMBER, not a UID, so the + # old `else` fallback flagged whichever message happened to occupy sequence + # position == the UID value. When the UID isn't present, fail safe (callers + # surface "Email not found") rather than touch an unrelated message. + if not _uid_exists(conn, uid): + return False op = "+FLAGS" if add else "-FLAGS" - if _uid_exists(conn, uid): - status, _ = conn.uid("STORE", _uid_bytes(uid), op, flag) - else: - status, _ = conn.store(_uid_bytes(uid), op, flag) + status, _ = conn.uid("STORE", _uid_bytes(uid), op, flag) return status == "OK" def _move_email_message(conn, uid: str, dest: str, role: str = "") -> bool: dest = _resolve_mail_folder(conn, dest, role or _folder_role_from_name(dest)) - if _uid_exists(conn, uid): - status, _ = conn.uid("MOVE", _uid_bytes(uid), _q(dest)) - if status == "OK": - return True - status, _ = conn.uid("COPY", _uid_bytes(uid), _q(dest)) - if status != "OK": - return False - status, _ = conn.uid("STORE", _uid_bytes(uid), "+FLAGS", "\\Deleted") - else: - status, _ = conn.copy(_uid_bytes(uid), _q(dest)) - if status != "OK": - return False - status, _ = conn.store(_uid_bytes(uid), "+FLAGS", "\\Deleted") + # copy()/store() are SEQUENCE-NUMBER commands; using them with a UID (the old + # `else` branch) copied + \Deleted-flagged the wrong message and then + # expunge() permanently removed it. There is no valid case where treating a + # UID as a sequence number is correct, so fail safe when the UID is absent. + if not _uid_exists(conn, uid): + return False + status, _ = conn.uid("MOVE", _uid_bytes(uid), _q(dest)) + if status == "OK": + return True + status, _ = conn.uid("COPY", _uid_bytes(uid), _q(dest)) + if status != "OK": + return False + status, _ = conn.uid("STORE", _uid_bytes(uid), "+FLAGS", "\\Deleted") if status == "OK": conn.expunge() return True diff --git a/routes/history/history_routes.py b/routes/history/history_routes.py index 2324ae286a..d0d45e8eb2 100644 --- a/routes/history/history_routes.py +++ b/routes/history/history_routes.py @@ -10,7 +10,9 @@ from core.models import ChatMessage from core.database import SessionLocal, ChatMessage as DbChatMessage, Session as DbSession +from src.auth_helpers import effective_user from src.topic_analyzer import analyze_topics +from src.upload_handler import reserve_message_upload_references from routes.session_routes import ( _message_role, _message_text, @@ -98,9 +100,29 @@ def _merge_continue_rows_to_delete(db_messages, db1, db2): return to_delete -def setup_history_routes(session_manager) -> APIRouter: +def setup_history_routes(session_manager, upload_handler=None) -> APIRouter: router = APIRouter(tags=["history"]) + def _reserve_message_uploads( + request: Request, + content: Any, + metadata: Any = None, + ) -> None: + try: + missing_id = reserve_message_upload_references( + upload_handler, + effective_user(request), + content, + metadata, + ) + except (TypeError, ValueError) as exc: + raise HTTPException(400, "Invalid message attachment metadata") from exc + if missing_id: + raise HTTPException( + 409, + f"Referenced upload is no longer available: {missing_id}", + ) + def _db_history_entry(m: DbChatMessage) -> Dict[str, Any]: entry = {"role": m.role, "content": _history_display_content(m.content)} meta = {} @@ -251,7 +273,9 @@ async def add_message(request: Request, session_id: str): content = body.get("content", "") if not content: raise HTTPException(400, "content is required") - msg = ChatMessage(role=role, content=content, metadata=body.get("metadata")) + metadata = body.get("metadata") + _reserve_message_uploads(request, content, metadata) + msg = ChatMessage(role=role, content=content, metadata=metadata) session_manager.add_message(session_id, msg) return {"status": "ok"} except KeyError: @@ -331,6 +355,8 @@ async def edit_message(request: Request, session_id: str): if not msg_id or content is None: raise HTTPException(400, "msg_id and content are required") + _reserve_message_uploads(request, content) + session = session_manager.get_session(session_id) db = SessionLocal() try: diff --git a/routes/note_routes.py b/routes/note_routes.py index 3d356f5c99..ec45072646 100644 --- a/routes/note_routes.py +++ b/routes/note_routes.py @@ -13,6 +13,7 @@ from core.middleware import INTERNAL_TOOL_USER from src.auth_helpers import require_user from src.constants import DATA_DIR +from src.upload_handler import reserve_upload_references from sqlalchemy.orm.attributes import flag_modified logger = logging.getLogger(__name__) @@ -479,7 +480,9 @@ def _smtp_send(): base = intg["base_url"].rstrip("/") topic = settings.get("reminder_ntfy_topic") or "reminders" ntfy_body = synthesis or note_body or title - hdrs = {"Title": title or "Reminder", "Priority": "high", "Tags": "bell"} + # ntfy Title is an ASCII HTTP header; sanitize Unicode and cap its length. + _clean_title = (title or "Reminder").encode("ascii", "replace").decode("ascii")[:200] + hdrs = {"Title": _clean_title, "Priority": "high", "Tags": "bell"} api_key = intg.get("api_key", "") if api_key: hdrs["Authorization"] = f"Bearer {api_key}" @@ -572,7 +575,7 @@ def _smtp_send(): # Router factory # --------------------------------------------------------------------------- -def setup_note_routes(task_scheduler=None): +def setup_note_routes(task_scheduler=None, upload_handler=None): # Expose the scheduler to module-level `dispatch_reminder` so reminders # can also push to the in-app notification queue (the polling system # turns each entry into a real browser Notification + the existing @@ -594,6 +597,11 @@ def _owner(request: Request) -> Optional[str]: # did not. return require_user(request) or None + def _reserve_note_uploads(owner: Optional[str], *values) -> None: + missing_id = reserve_upload_references(upload_handler, owner, *values) + if missing_id: + raise HTTPException(409, f"Referenced upload is no longer available: {missing_id}") + def _is_admin_or_single_user(request: Request, user: str | None) -> bool: if user == INTERNAL_TOOL_USER: return True @@ -643,6 +651,13 @@ def list_notes( @router.post("") def create_note(request: Request, body: NoteCreate): user = _owner(request) + _reserve_note_uploads( + user, + body.image_url, + body.color, + body.content, + json.dumps(body.items) if body.items is not None else None, + ) db = SessionLocal() try: note = Note( @@ -700,6 +715,13 @@ def update_note(request: Request, note_id: str, body: NoteUpdate): if user is not None and note.owner != user: raise HTTPException(404, "Note not found") + _reserve_note_uploads( + user, + body.image_url, + body.color, + body.content, + json.dumps(body.items) if body.items is not None else None, + ) if body.title is not None: note.title = body.title if body.content is not None: diff --git a/routes/session_routes.py b/routes/session_routes.py index 2d3543e369..2d8a6d87f3 100644 --- a/routes/session_routes.py +++ b/routes/session_routes.py @@ -13,6 +13,7 @@ from core.database import Session as DbSession, SessionLocal, Document, GalleryImage, utcnow_naive from src.auth_helpers import effective_user, _auth_disabled, owner_filter from src.session_actions import is_session_recently_active +from src.upload_handler import reserve_message_upload_references def _sanitize_export_filename(name: str) -> str: @@ -203,7 +204,12 @@ def _pick_endpoint_for_sort(owner=None): return url, model, headers return None, None, None -def setup_session_routes(session_manager: SessionManager, config: dict, webhook_manager=None): +def setup_session_routes( + session_manager: SessionManager, + config: dict, + webhook_manager=None, + upload_handler=None, +): """Setup session routes with the provided manager and config""" REQUEST_TIMEOUT = config.get("REQUEST_TIMEOUT", 20) @@ -537,6 +543,22 @@ async def inject_messages(request: Request, sid: str): body = await request.json() messages = body.get("messages", []) from core.models import ChatMessage + owner = effective_user(request) + try: + for message in messages: + missing_id = reserve_message_upload_references( + upload_handler, + owner, + message.get("content"), + message.get("metadata"), + ) + if missing_id: + raise HTTPException( + 409, + f"Referenced upload is no longer available: {missing_id}", + ) + except (AttributeError, TypeError, ValueError) as exc: + raise HTTPException(400, "Invalid message attachment metadata") from exc for m in messages: sess.add_message(ChatMessage(m["role"], m["content"], metadata=m.get("metadata"))) session_manager.save_sessions() diff --git a/routes/upload_routes.py b/routes/upload_routes.py index 1bd402ac8a..fb702e45a7 100644 --- a/routes/upload_routes.py +++ b/routes/upload_routes.py @@ -10,16 +10,140 @@ from typing import List, Optional import logging from core.middleware import require_admin -from core.database import SessionLocal, GalleryImage, Session as DbSession +from core.database import ( + SessionLocal, + ChatMessage as DbChatMessage, + CalendarCal, + CalendarEvent, + Document, + DocumentVersion, + GalleryImage, + Note, + Session as DbSession, +) from src.auth_helpers import effective_user +from src.attachment_refs import attachment_refs_from_metadata from src.constants import GENERATED_IMAGES_DIR -from src.upload_handler import count_recent_uploads +from src.upload_handler import ( + UploadCleanupSafetyError, + count_recent_uploads, + extract_upload_ids, +) logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/upload", tags=["upload"]) UPLOAD_RESPONSE_HEADERS = {"X-Content-Type-Options": "nosniff"} +def _upload_ids_from_persisted_text(value: object) -> set[str]: + """Return canonical upload IDs embedded in persisted text. + + This covers attachment reference lines/URIs and the PDF source markers + stored by the document editor. False positives are intentionally + conservative: retaining an extra upload is safer than deleting referenced + bytes. + """ + return extract_upload_ids(value) + + +def _upload_ids_from_message_metadata(raw_metadata: object) -> set[str]: + """Extract attachment IDs from a persisted chat metadata JSON value. + + Malformed metadata raises instead of being treated as an empty reference + set. The admin cleanup route catches that failure and aborts cleanup. + """ + if raw_metadata in (None, ""): + return set() + if isinstance(raw_metadata, str): + metadata = json.loads(raw_metadata) + else: + metadata = raw_metadata + if not isinstance(metadata, dict): + raise ValueError("chat message metadata must be a JSON object") + + attachments = metadata.get("attachments") + if attachments is not None: + if not isinstance(attachments, list) or any( + not isinstance(item, dict) for item in attachments + ): + raise ValueError("chat message attachments metadata is malformed") + + ids = { + str(ref["attachment_id"]) + for ref in attachment_refs_from_metadata(metadata) + if ref.get("attachment_id") + } + # Preserve canonical IDs even in older metadata shapes not normalized by + # attachment_refs_from_metadata(). + ids.update(_upload_ids_from_persisted_text(json.dumps(metadata))) + return ids + + +def _collect_persisted_upload_references() -> tuple[set[str], set[str]]: + """Collect upload IDs/hashes still referenced by durable application data. + + The caller must treat any exception as an incomplete scan and fail closed. + There is no distinct artifact table in the current schema; artifact-like + attachment references persisted in chat/document text are covered by the + canonical-ID scan. + """ + referenced_ids: set[str] = set() + referenced_hashes: set[str] = set() + db = SessionLocal() + try: + for content, raw_metadata in db.query( + DbChatMessage.content, + DbChatMessage.meta_data, + ).yield_per(500): + referenced_ids.update(_upload_ids_from_persisted_text(content)) + referenced_ids.update(_upload_ids_from_message_metadata(raw_metadata)) + + for (content,) in db.query(Document.current_content).yield_per(500): + referenced_ids.update(_upload_ids_from_persisted_text(content)) + + for (content,) in db.query(DocumentVersion.content).yield_per(500): + referenced_ids.update(_upload_ids_from_persisted_text(content)) + + for filename, file_hash in db.query( + GalleryImage.filename, + GalleryImage.file_hash, + ).yield_per(500): + referenced_ids.update(_upload_ids_from_persisted_text(filename)) + if file_hash: + referenced_hashes.add(str(file_hash)) + + for image_url, color, content, items in db.query( + Note.image_url, + Note.color, + Note.content, + Note.items, + ).yield_per(500): + for value in (image_url, color, content, items): + referenced_ids.update(_upload_ids_from_persisted_text(value)) + + for (color,) in db.query(CalendarCal.color).yield_per(500): + referenced_ids.update(_upload_ids_from_persisted_text(color)) + + for color, description, location in db.query( + CalendarEvent.color, + CalendarEvent.description, + CalendarEvent.location, + ).yield_per(500): + for value in (color, description, location): + referenced_ids.update(_upload_ids_from_persisted_text(value)) + + return referenced_ids, referenced_hashes + finally: + db.close() + + +def _run_reference_safe_cleanup(upload_handler) -> int: + referenced_ids, referenced_hashes = _collect_persisted_upload_references() + return upload_handler.cleanup_old_uploads( + referenced_upload_ids=referenced_ids, + referenced_upload_hashes=referenced_hashes, + ) + def setup_upload_routes(upload_handler): """Setup upload routes with the provided handler""" @@ -172,7 +296,9 @@ async def api_upload( "mime": meta["mime"], "size": meta["size"], "hash": meta["hash"], + "checksum_sha256": meta.get("checksum_sha256") or meta["hash"], "uploaded_at": meta["uploaded_at"], + "created_at": meta.get("created_at") or meta["uploaded_at"], "width": meta.get("width"), "height": meta.get("height"), "is_duplicate": meta.get("is_duplicate", False) @@ -195,7 +321,23 @@ async def api_upload( async def manual_cleanup(request: Request): """Manually trigger cleanup of old uploads.""" require_admin(request) - cleaned_count = upload_handler.cleanup_old_uploads() + try: + cleaned_count = await asyncio.to_thread( + _run_reference_safe_cleanup, + upload_handler, + ) + except UploadCleanupSafetyError: + logger.exception("Upload cleanup aborted because index safety checks failed") + raise HTTPException( + 503, + "Upload cleanup aborted because upload index integrity could not be verified", + ) + except Exception: + logger.exception("Upload cleanup skipped because reference discovery failed") + raise HTTPException( + 503, + "Upload cleanup skipped because persisted references could not be verified", + ) return {"status": "success", "files_cleaned": cleaned_count} @router.get("/stats") diff --git a/scripts/odysseus-backup b/scripts/odysseus-backup index b0f3120746..9709ed6b53 100755 --- a/scripts/odysseus-backup +++ b/scripts/odysseus-backup @@ -133,18 +133,32 @@ def cmd_list(args): emit([], args) return entries = [] - for p in sorted(_BACKUP_DIR.iterdir(), key=lambda x: x.stat().st_mtime, reverse=True): - if not p.is_file(): - continue - entries.append({ - "path": str(p), - "name": p.name, - "bytes": p.stat().st_size, - "modified": datetime.fromtimestamp(p.stat().st_mtime).isoformat(), - }) + for p in _BACKUP_DIR.iterdir(): + entry = _backup_entry(p) + if entry is not None: + entries.append(entry) + entries.sort(key=lambda entry: entry["_mtime"], reverse=True) + for entry in entries: + entry.pop("_mtime", None) emit(entries, args) +def _backup_entry(p): + try: + if not p.is_file(): + return None + st = p.stat() + except OSError: + return None + return { + "path": str(p), + "name": p.name, + "bytes": st.st_size, + "modified": datetime.fromtimestamp(st.st_mtime).isoformat(), + "_mtime": st.st_mtime, + } + + def cmd_verify(args): """Open the tarball read-only and walk its members — confirms integrity without extracting anything.""" diff --git a/scripts/odysseus-memory b/scripts/odysseus-memory index 1a4f8a0337..04ef67894f 100755 --- a/scripts/odysseus-memory +++ b/scripts/odysseus-memory @@ -90,7 +90,7 @@ def cmd_add(args): # add_entry doesn't save by default — the call in chat does it # after dedup checks. Persist here so a one-shot CLI add sticks. all_entries = _manager().load_all() - if not any(e.get("id") == entry.get("id") for e in all_entries): + if not any(isinstance(e, dict) and e.get("id") == entry.get("id") for e in all_entries): all_entries.append(entry) _manager().save(all_entries) emit(entry, args) diff --git a/services/hwfit/models.py b/services/hwfit/models.py index 6f2bb00e7a..c042c9462a 100644 --- a/services/hwfit/models.py +++ b/services/hwfit/models.py @@ -145,7 +145,7 @@ def is_prequantized(model): or re.search(r"(^|[-_/])fp8($|[-_/\s])", text) is not None or (not (model.get("is_gguf") or model.get("gguf_sources")) and re.search(r"(^|[-_/])(?:int)?8bit($|[-_/\s])", text) is not None) or any(x in text for x in ("awq", "gptq", "mlx")) - or any(q.startswith(p) for p in PREQUANTIZED_PREFIXES) + or any(isinstance(q, str) and q.startswith(p) for p in PREQUANTIZED_PREFIXES) ) @@ -155,7 +155,7 @@ def params_b(model): return raw / 1_000_000_000.0 pc = model.get("parameter_count", "") - if pc: + if isinstance(pc, str) and pc: pc = pc.strip().upper() m = re.match(r"^([\d.]+)\s*([BKMGT]?)$", pc) if m: diff --git a/services/memory/skill_importer.py b/services/memory/skill_importer.py index 65f4b21c0a..2f0d7ab326 100644 --- a/services/memory/skill_importer.py +++ b/services/memory/skill_importer.py @@ -6,7 +6,7 @@ import re from dataclasses import dataclass from typing import Dict, List, Optional, Tuple -from urllib.parse import quote, urlparse +from urllib.parse import quote, urljoin, urlparse import httpx @@ -68,6 +68,52 @@ def _is_text_file(name: str) -> bool: return any(low.endswith(s) for s in ALLOWED_SUFFIXES) +# Max redirect hops to follow manually while re-validating each one. +_MAX_FETCH_REDIRECTS = 5 + + +def _check_fetch_url(url: str) -> None: + """SSRF guard for skill-import fetches (defense-in-depth). + + Skill bundles only ever come from public GitHub, never an internal + address, so block private/loopback/link-local targets on every hop — + matching the hardened web-fetch path in + ``services/search/content.py:_get_public_url`` rather than the lenient + default used for admin-configured model endpoints. + """ + ok, reason = check_outbound_url(url, block_private=True) + if not ok: + raise SkillImportError(reason) + + +def _get_checked( + url: str, + *, + headers: Optional[dict] = None, + timeout: float = 30.0, +) -> httpx.Response: + """GET that follows redirects manually, re-running the SSRF guard per hop. + + ``httpx``'s ``follow_redirects=True`` validates only the initial URL, so a + ``3xx`` to an internal address (``169.254.169.254``, ``127.0.0.1``, …) would + still be connected to before any post-hoc host check. Following redirects by + hand lets us re-validate every hop, closing that blind-SSRF gap. + """ + current = url + with httpx.Client(follow_redirects=False, timeout=timeout) as client: + for _ in range(_MAX_FETCH_REDIRECTS + 1): + _check_fetch_url(current) + r = client.get(current, headers=headers) + if r.status_code in (301, 302, 303, 307, 308): + location = r.headers.get("location") + if not location: + return r + current = urljoin(str(r.url), location) + continue + return r + raise SkillImportError("too many redirects while fetching skill bundle") + + def parse_skill_source(url: str) -> ResolvedSource: """Normalize skills.sh / GitHub web URLs into owner/repo/ref/path.""" raw = (url or "").strip() @@ -76,22 +122,18 @@ def parse_skill_source(url: str) -> ResolvedSource: # skills.sh often links to GitHub; try to unwrap ?url= or redirect target later. if "skills.sh" in raw and "github.com" not in raw: - ok, reason = check_outbound_url(raw) - if not ok: - raise SkillImportError(reason) - with httpx.Client(follow_redirects=True, timeout=20.0) as client: - r = client.get(raw) - if r.status_code >= 400: - raise _github_response_error(r) - final = str(r.url) - _assert_github_url(final, context="redirect target") - # Page may embed a github link; prefer final URL if redirected. - if "github.com" in final: - raw = final - else: - m = re.search(r"https?://github\.com/[^\s\"')]+", r.text or "") - if m: - raw = m.group(0).rstrip(".,)") + r = _get_checked(raw, timeout=20.0) + if r.status_code >= 400: + raise _github_response_error(r) + final = str(r.url) + _assert_github_url(final, context="redirect target") + # Page may embed a github link; prefer final URL if redirected. + if "github.com" in final: + raw = final + else: + m = re.search(r"https?://github\.com/[^\s\"')]+", r.text or "") + if m: + raw = m.group(0).rstrip(".,)") parsed = urlparse(raw) host = _github_host(raw) @@ -165,17 +207,13 @@ def _github_response_error(response: httpx.Response) -> SkillImportError: def _fetch_bytes(url: str) -> bytes: - ok, reason = check_outbound_url(url) - if not ok: - raise SkillImportError(reason) - with httpx.Client(follow_redirects=True, timeout=30.0) as client: - r = client.get(url, headers={"Accept": "application/vnd.github+json"}) - if r.status_code >= 400: - raise _github_response_error(r) - _assert_github_url(str(r.url), context="redirect target") - if len(r.content) > MAX_FILE_BYTES: - raise SkillImportError(f"file too large: {url}") - return r.content + r = _get_checked(url, headers={"Accept": "application/vnd.github+json"}, timeout=30.0) + if r.status_code >= 400: + raise _github_response_error(r) + _assert_github_url(str(r.url), context="redirect target") + if len(r.content) > MAX_FILE_BYTES: + raise SkillImportError(f"file too large: {url}") + return r.content def _fetch_text(url: str) -> str: @@ -190,15 +228,11 @@ def _list_github_dir(src: ResolvedSource, rel_dir: str, out: Dict[str, str], *, if depth > 4 or len(out) >= MAX_FILES: return url = _api_contents_url(src, rel_dir) - ok, reason = check_outbound_url(url) - if not ok: - raise SkillImportError(reason) - with httpx.Client(follow_redirects=True, timeout=30.0) as client: - r = client.get(url, headers={"Accept": "application/vnd.github+json"}) - if r.status_code >= 400: - raise _github_response_error(r) - _assert_github_url(str(r.url), context="redirect target") - entries = r.json() + r = _get_checked(url, headers={"Accept": "application/vnd.github+json"}, timeout=30.0) + if r.status_code >= 400: + raise _github_response_error(r) + _assert_github_url(str(r.url), context="redirect target") + entries = r.json() if not isinstance(entries, list): raise SkillImportError("expected a directory on GitHub") total = sum(len(v.encode("utf-8")) for v in out.values()) diff --git a/services/tts/tts_service.py b/services/tts/tts_service.py index e724434cb6..2120d7720f 100644 --- a/services/tts/tts_service.py +++ b/services/tts/tts_service.py @@ -68,7 +68,7 @@ def available(self) -> bool: if provider == "local": kokoro = self._get_kokoro() return kokoro is not None and kokoro.available - if provider.startswith("endpoint:"): + if isinstance(provider, str) and provider.startswith("endpoint:"): return True # assume reachable; errors surface at synthesis time return False diff --git a/src/agent_loop.py b/src/agent_loop.py index ebc2a99a62..6f7dde6054 100644 --- a/src/agent_loop.py +++ b/src/agent_loop.py @@ -24,7 +24,7 @@ from src.settings import get_setting from src.prompt_security import untrusted_context_message from src.tool_security import blocked_tools_for_owner, plan_mode_disabled_tools -from src.tool_policy import GUIDE_ONLY_DIRECTIVE, ToolPolicy +from src.tool_policy import GUIDE_ONLY_DIRECTIVE, WEB_TOOL_NAMES, ToolPolicy from src.tool_utils import _truncate, get_mcp_manager from src.agent_tools import ( parse_tool_blocks, @@ -321,7 +321,7 @@ def _load_mcp_disabled_map() -> Dict[str, set]: } _DOMAIN_TOOL_MAP = { - "web": {"web_search", "web_fetch"}, + "web": set(WEB_TOOL_NAMES), "documents": {"create_document", "edit_document", "update_document", "suggest_document", "manage_documents"}, "email": {"list_email_accounts", "list_emails", "read_email", "send_email", "reply_to_email", "bulk_email", "archive_email", "delete_email", "mark_email_read", "resolve_contact", "manage_contact"}, "cookbook": {"download_model", "serve_model", "serve_preset", "list_serve_presets", "list_served_models", "stop_served_model", "tail_serve_output", "list_downloads", "cancel_download", "search_hf_models", "list_cached_models", "list_cookbook_servers", "adopt_served_model"}, @@ -2847,13 +2847,12 @@ async def stream_agent_loop( if "email" in (_intent.get("domains") or set()): _relevant_tools.add("ui_control") if "web" in (_intent.get("domains") or set()): - _relevant_tools.update({"web_search", "web_fetch"}) - _removed_web_blocks = sorted({"web_search", "web_fetch"} & disabled_tools) - if _removed_web_blocks: - disabled_tools.difference_update({"web_search", "web_fetch"}) + _relevant_tools.update(WEB_TOOL_NAMES) + _blocked_web_tools = sorted(WEB_TOOL_NAMES & disabled_tools) + if _blocked_web_tools: logger.info( - "[agent-intent] web turn forced search tools enabled; removed disabled=%s", - _removed_web_blocks, + "[agent-intent] web domain selected but search tools remain disabled=%s", + _blocked_web_tools, ) if "ui" in (_intent.get("domains") or set()): _relevant_tools.add("ui_control") @@ -2887,9 +2886,9 @@ async def stream_agent_loop( _relevant_tools = set(ALWAYS_AVAILABLE) _relevant_tools.update({"read_file", "grep", "ls", "manage_documents"}) - # Per-request forced tools are stronger than retrieval. Search toggles and - # explicit lookup turns must make web tools visible even when tool RAG - # misses them; route-level disabled_tools decides what else is allowed. + # Per-request forced tools are stronger than retrieval. Explicit search + # settings make web tools visible even when tool RAG misses them; + # route-level disabled_tools decides what remains allowed. if not guide_only and forced_tools: forced_set = {t for t in forced_tools if t not in disabled_tools} if _relevant_tools is None: @@ -3829,9 +3828,8 @@ async def stream_agent_loop( and _intent_match is not None and len(_intent_text) < 400 and "```" not in _intent_text - and _intent_nudge_count < _MAX_INTENT_NUDGES ) - if _looks_like_promise: + if _looks_like_promise and _intent_nudge_count < _MAX_INTENT_NUDGES: _intent_nudge_count += 1 _matched_phrase = _intent_match.group(0).strip() logger.info(f"[agent] intent-without-action nudge #{_intent_nudge_count} on round {round_num}: {_matched_phrase!r}") @@ -3860,6 +3858,31 @@ async def stream_agent_loop( # Visible signal in the stream so the user knows we caught it. yield f'data: {json.dumps({"type": "agent_step", "round": round_num + 1})}\n\n' continue + if _looks_like_promise: + _matched_phrase = _intent_match.group(0).strip() + _guard_message = ( + "The agent stopped because it repeatedly announced a tool " + "action without making the tool call." + ) + logger.warning( + "[agent] intent-without-action guard exhausted on round %d after %d nudges: %r", + round_num, + _intent_nudge_count, + _matched_phrase, + ) + yield ( + "data: " + + json.dumps({ + "type": "intent_nudge_exhausted", + "reason": "intent_without_action_nudge_cap", + "message": _guard_message, + "round": round_num, + "nudges": _intent_nudge_count, + "matched": _matched_phrase, + }) + + "\n\n" + ) + break break # no tools — done # ── Loop-breaker (Terminus-style stall detector) ────────────── @@ -3896,6 +3919,21 @@ async def stream_agent_loop( reason = (f"calling {_runaway} with identical arguments over and over" if _runaway else "repeating the same tool calls without new progress") logger.warning(f"[agent] loop-breaker tripped on round {round_num} ({reason}); sig={_sig[:80]!r}") + yield ( + "data: " + + json.dumps({ + "type": "loop_breaker_triggered", + "reason": "loop_breaker_stall", + "message": ( + "The loop-breaker detected repeated tool calls without " + "new progress, so the agent is being forced to stop " + "using tools and give its best final answer." + ), + "round": round_num, + "detail": reason, + }) + + "\n\n" + ) # The model has been executing tools, so its results are already # in context. Force ONE tool-free round to converge: write the # answer from what it has, or state plainly what's blocking it. @@ -4014,16 +4052,31 @@ async def _run_tool(): await _progress_q.put(None) _tool_task = asyncio.create_task(_run_tool()) - # Drain progress events as they arrive — block until the - # next event OR the tool finishes (sentinel = None). - while True: - evt = await _progress_q.get() - if evt is None: - break - yield ( - f'data: {json.dumps({"type": "tool_progress", "tool": block.tool_type, "round": round_num, **evt})}\n\n' - ) - desc, result = await _tool_task + try: + # Drain progress events as they arrive — block until the + # next event OR the tool finishes (sentinel = None). + while True: + evt = await _progress_q.get() + if evt is None: + break + yield ( + f'data: {json.dumps({"type": "tool_progress", "tool": block.tool_type, "round": round_num, **evt})}\n\n' + ) + desc, result = await _tool_task + finally: + # If the SSE client disconnects (or this generator is + # otherwise closed) while we're awaiting a progress event + # above, GeneratorExit is thrown in right here and the + # `await _tool_task` on the line above never runs — the + # task (and any subprocess execute_tool_block spawned for + # bash/python tools) would otherwise keep running + # orphaned with nothing left to await or cancel it. + if not _tool_task.done(): + _tool_task.cancel() + try: + await _tool_task + except (asyncio.CancelledError, Exception): + pass # A skill the model just loaded can prescribe tools that weren't # RAG-selected this turn (declared via requires_toolsets in its diff --git a/src/agent_tools/document_tools.py b/src/agent_tools/document_tools.py index 2ab8659d30..65ee0461ec 100644 --- a/src/agent_tools/document_tools.py +++ b/src/agent_tools/document_tools.py @@ -2,10 +2,16 @@ import logging import re from src.constants import MAX_READ_CHARS -from src.tool_utils import _parse_tool_args +from src.tool_utils import _parse_tool_args, get_upload_handler +from src.upload_handler import reserve_upload_references logger = logging.getLogger(__name__) + +def _missing_document_upload(owner: Optional[str], content: Any) -> Optional[str]: + """Reserve explicit upload URLs before an agent persists document text.""" + return reserve_upload_references(get_upload_handler(), owner, content) + # --------------------------------------------------------------------------- # Active document state # --------------------------------------------------------------------------- @@ -384,6 +390,13 @@ async def execute(self, content: str, ctx: dict) -> dict: return {"error": "Cannot create document in another user's session"} _owner = _sess.owner if _sess else None + missing_id = _missing_document_upload(_owner, content) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } + doc = Document( id=doc_id, session_id=session_id, @@ -455,6 +468,13 @@ async def execute(self, content: str, ctx: dict) -> Dict: if is_email_doc: doc.language = "email" + missing_id = _missing_document_upload(owner, new_content) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } + if not is_email_doc and _pdf_source_upload_id(doc.current_content or ""): return _create_pdf_text_derivative( db, @@ -532,6 +552,12 @@ async def execute(self, content: str, ctx: dict) -> Dict: applied = 1 skipped = max(0, len(edits) - 1) doc.language = "email" + missing_id = _missing_document_upload(owner, updated_content) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } new_ver = doc.version_count + 1 ver = DocumentVersion( id=str(uuid.uuid4()), @@ -584,6 +610,13 @@ async def execute(self, content: str, ctx: dict) -> Dict: if applied == 0: return {"error": f"No edits applied — none of the FIND blocks matched the document content (skipped {skipped})"} + missing_id = _missing_document_upload(owner, updated_content) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } + if _pdf_source_upload_id(doc.current_content or ""): return _create_pdf_text_derivative( db, diff --git a/src/app_initializer.py b/src/app_initializer.py index b438b2b173..1b29f06d26 100644 --- a/src/app_initializer.py +++ b/src/app_initializer.py @@ -21,6 +21,7 @@ from src.chat_handler import ChatHandler from src.research_handler import ResearchHandler from src.upload_handler import UploadHandler +from src.tool_utils import set_upload_handler from src.search import update_search_config logger = logging.getLogger(__name__) @@ -49,6 +50,8 @@ def initialize_managers(base_dir: str, rag_manager=None) -> Dict[str, Any]: session_manager = SessionManager(SESSIONS_FILE) set_session_manager(session_manager) # Enable Session.add_message() persistence upload_handler = UploadHandler(base_dir, UPLOAD_DIR) + session_manager.upload_handler = upload_handler + set_upload_handler(upload_handler) personal_docs_manager = PersonalDocsManager(PERSONAL_DIR, rag_manager) api_key_manager = APIKeyManager(DATA_DIR) preset_manager = PresetManager(DATA_DIR) diff --git a/src/attachment_refs.py b/src/attachment_refs.py new file mode 100644 index 0000000000..054bea54b9 --- /dev/null +++ b/src/attachment_refs.py @@ -0,0 +1,164 @@ +"""Attachment reference helpers for chat storage and tool manifests. + +Live model calls may need provider-specific data URLs for the current turn. +Persisted history and search indexes should keep stable upload references and +human-readable text instead of duplicating raw media bytes. +""" + +from __future__ import annotations + +import json +import re +from typing import Any, Iterable + + +DATA_URL_RE = re.compile( + r"data:[^;,\s\"']+;base64,[A-Za-z0-9+/=]+", + re.IGNORECASE, +) + +MEDIA_BLOCK_TYPES = { + "image", + "image_url", + "input_image", + "audio", + "input_audio", + "file", +} + + +def strip_inline_data_urls(text: str) -> str: + """Replace inline data URLs with a compact marker.""" + if not isinstance(text, str) or ";base64," not in text: + return text + return DATA_URL_RE.sub("[inline media omitted from persisted history]", text) + + +def attachment_ref(info: dict[str, Any]) -> dict[str, Any]: + """Return the stable attachment reference shape used outside raw uploads.""" + upload_id = str(info.get("id") or info.get("attachment_id") or "").strip() + try: + size = int(info.get("size") or 0) + except (TypeError, ValueError): + size = 0 + ref = { + "type": "attachment_ref", + "attachment_id": upload_id, + "name": info.get("name") or info.get("original_name") or upload_id, + "mime": info.get("mime") or "application/octet-stream", + "size": size, + } + checksum = info.get("checksum_sha256") or info.get("sha256") or info.get("hash") + if checksum: + ref["checksum_sha256"] = checksum + created_at = info.get("created_at") or info.get("uploaded_at") + if created_at: + ref["created_at"] = created_at + for key in ("width", "height", "vision", "vision_model", "gallery_id"): + value = info.get(key) + if value is not None: + ref[key] = value + return ref + + +def attachment_refs_from_metadata(metadata: dict[str, Any] | None) -> list[dict[str, Any]]: + """Extract attachment refs from message metadata.""" + attachments = (metadata or {}).get("attachments") or [] + if not isinstance(attachments, list): + return [] + refs: list[dict[str, Any]] = [] + for item in attachments: + if isinstance(item, dict): + ref = attachment_ref(item) + if ref.get("attachment_id"): + refs.append(ref) + return refs + + +def _ref_line(ref: dict[str, Any]) -> str: + parts = [f"Attachment: {ref.get('name') or ref.get('attachment_id') or 'upload'}"] + if ref.get("attachment_id"): + parts.append(f"id={ref['attachment_id']}") + if ref.get("mime"): + parts.append(f"mime={ref['mime']}") + if ref.get("size"): + parts.append(f"size={ref['size']} bytes") + if ref.get("checksum_sha256"): + parts.append(f"sha256={ref['checksum_sha256']}") + line = "[" + " | ".join(parts) + "]" + if ref.get("vision"): + line += f"\n[Attachment description: {str(ref['vision']).strip()}]" + return line + + +def _text_from_blocks(blocks: Iterable[Any]) -> str: + lines: list[str] = [] + omitted_media = 0 + for block in blocks: + if isinstance(block, str): + lines.append(strip_inline_data_urls(block)) + continue + if not isinstance(block, dict): + continue + block_type = block.get("type") + if block_type == "text": + text = block.get("text") + if isinstance(text, str) and text: + lines.append(strip_inline_data_urls(text)) + elif block_type == "attachment_ref": + lines.append(_ref_line(block)) + elif block_type in MEDIA_BLOCK_TYPES: + omitted_media += 1 + else: + try: + encoded = json.dumps(block, ensure_ascii=True, sort_keys=True) + except TypeError: + encoded = str(block) + lines.append(strip_inline_data_urls(encoded)) + if omitted_media: + plural = "s" if omitted_media != 1 else "" + lines.append(f"[{omitted_media} inline media payload{plural} omitted]") + return "\n".join(line for line in lines if line).strip() + + +def persistable_message_content( + content: Any, + metadata: dict[str, Any] | None = None, +) -> str: + """Return content safe for DB persistence and FTS indexing. + + Multimodal provider blocks are collapsed to readable text plus stable + attachment reference lines from metadata. This avoids storing base64 media + in ``chat_messages.content`` while preserving enough context for reloads, + search, and later turns. + """ + if isinstance(content, list): + text = _text_from_blocks(content) + refs = attachment_refs_from_metadata(metadata) + ref_lines = [_ref_line(ref) for ref in refs] + if ref_lines: + text = "\n".join([part for part in (text, "\n".join(ref_lines)) if part]).strip() + return text + if isinstance(content, str): + return strip_inline_data_urls(content) + try: + return strip_inline_data_urls(json.dumps(content, ensure_ascii=True, sort_keys=True)) + except TypeError: + return strip_inline_data_urls(str(content)) + + +def search_index_text(content: Any) -> str: + """Best-effort searchable text for legacy stored content.""" + if isinstance(content, str): + raw = content.strip() + if raw.startswith("[") and '"type"' in raw: + try: + parsed = json.loads(content) + except (TypeError, ValueError): + parsed = None + if isinstance(parsed, list): + return _text_from_blocks(parsed) + return strip_inline_data_urls(content) + if isinstance(content, list): + return _text_from_blocks(content) + return persistable_message_content(content) diff --git a/src/builtin_actions.py b/src/builtin_actions.py index 727d2699d1..ca5e5158ff 100644 --- a/src/builtin_actions.py +++ b/src/builtin_actions.py @@ -1125,14 +1125,14 @@ def _pull_headers(): conn = _imap_connect(None, owner=owner) try: conn.select("INBOX", readonly=True) - status, data = conn.search(None, "ALL") + status, data = conn.uid("SEARCH", None, "ALL") if status != "OK" or not data or not data[0]: return results uids = data[0].split()[-300:][::-1] # newest 300 for uid in uids: try: - st, msg_data = conn.fetch( - uid, "(BODY.PEEK[HEADER.FIELDS (FROM)])" + st, msg_data = conn.uid( + "FETCH", uid, "(BODY.PEEK[HEADER.FIELDS (FROM)])" ) if st != "OK" or not msg_data or not msg_data[0]: continue @@ -1214,7 +1214,7 @@ def _fetch_bodies(_msgs): conn2.select("INBOX", readonly=True) for mm in _msgs: try: - st, data = conn2.fetch(mm["uid"], "(BODY.PEEK[TEXT])") + st, data = conn2.uid("FETCH", mm["uid"], "(BODY.PEEK[TEXT])") if st != "OK" or not data or not data[0]: continue raw = data[0][1] if isinstance(data[0], tuple) else None @@ -1356,13 +1356,13 @@ async def action_daily_brief(owner: str, **kwargs) -> Tuple[str, bool]: conn = _imap_connect(None) try: conn.select("INBOX", readonly=True) - status, data = conn.search(None, "UNSEEN") + status, data = conn.uid("SEARCH", None, "UNSEEN") uids = (data[0].split() if status == "OK" and data and data[0] else []) unread_count = len(uids) # Grab headers for the most recent 5 unread (UIDs increase with arrival) for uid in uids[-5:][::-1]: try: - _, msg_data = conn.fetch(uid, "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT)])") + _, msg_data = conn.uid("FETCH", uid, "(BODY.PEEK[HEADER.FIELDS (FROM SUBJECT)])") if not msg_data or not msg_data[0]: continue hdr = msg_data[0][1] if isinstance(msg_data[0], tuple) else msg_data[0] diff --git a/src/builtin_mcp.py b/src/builtin_mcp.py index 2a4b748eee..b777e99a4b 100644 --- a/src/builtin_mcp.py +++ b/src/builtin_mcp.py @@ -104,6 +104,21 @@ def _spawn_bg(coro) -> asyncio.Task: return task +def builtin_python_env(base_dir: str) -> dict[str, str]: + """Environment for built-in Python MCP subprocesses. + + The app root must be importable so mcp_servers can import local modules, but + replacing PYTHONPATH entirely hides site-packages in container/dev launches + that rely on PYTHONPATH for their active environment. + """ + existing = os.environ.get("PYTHONPATH", "") + parts = [base_dir] + for item in existing.split(os.pathsep): + if item and item not in parts: + parts.append(item) + return {"PYTHONPATH": os.pathsep.join(parts)} + + async def register_builtin_servers(mcp_manager): """Connect all built-in MCP servers to the manager.""" if MCP_DISABLED: @@ -121,7 +136,7 @@ async def _connect_python_server(server_id: str, script_path: str, name: str): transport="stdio", command=python, args=[script_path], - env={"PYTHONPATH": base_dir}, + env=builtin_python_env(base_dir), ) if ok: logger.info(f"Built-in MCP server registered: {name}") diff --git a/src/caldav_sync.py b/src/caldav_sync.py index f91ebc1a0a..47fb3333b3 100644 --- a/src/caldav_sync.py +++ b/src/caldav_sync.py @@ -280,214 +280,216 @@ def _sync_blocking(owner: str, url: str, username: str, password: str, account_i result = {"calendars": 0, "events": 0, "deleted": 0, "errors": []} client = _build_dav_client(url, username, password) - - # Discovery: try principal → calendars first; if the server doesn't - # support discovery (or the URL points directly at a calendar), fall - # back to treating the URL as a single calendar. - calendars = [] try: - principal = client.principal() - calendars = principal.calendars() - except (AuthorizationError, NotFoundError) as e: - result["errors"].append(f"Discovery failed: {e}") - return result - except Exception as e: - logger.info(f"CalDAV principal discovery failed, trying URL as calendar: {e}") + # Discovery: try principal → calendars first; if the server doesn't + # support discovery (or the URL points directly at a calendar), fall + # back to treating the URL as a single calendar. + calendars = [] try: - calendars = [_open_url_as_calendar(client, url)] - except Exception as e2: - result["errors"].append(f"Could not open URL as calendar: {e2}") - return result - - if not calendars: - try: - calendars = [_open_url_as_calendar(client, url)] + principal = client.principal() + calendars = principal.calendars() + except (AuthorizationError, NotFoundError) as e: + result["errors"].append(f"Discovery failed: {e}") + return result # outer finally will call client.close() except Exception as e: - result["errors"].append(f"No calendars and URL fallback failed: {e}") - return result - - start = datetime.utcnow() - timedelta(days=_LOOKBACK_DAYS) - end = datetime.utcnow() + timedelta(days=_LOOKAHEAD_DAYS) + logger.info(f"CalDAV principal discovery failed, trying URL as calendar: {e}") + try: + calendars = [_open_url_as_calendar(client, url)] + except Exception as e2: + result["errors"].append(f"Could not open URL as calendar: {e2}") + return result # outer finally will call client.close() - db = SessionLocal() - try: - for remote_cal in calendars: + if not calendars: try: - remote_url = str(remote_cal.url) - cal_id = _stable_cal_id(remote_url, owner=owner, account_id=account_id) - display_name = (remote_cal.name or "").strip() or "CalDAV" - - local_cal = db.query(CalendarCal).filter( - CalendarCal.id == cal_id, - CalendarCal.owner == owner, - ).first() - if not local_cal: - local_cal = CalendarCal( - id=cal_id, - owner=owner, - name=display_name, - color="#5b8abf", - source="caldav", - account_id=account_id or None, - caldav_base_url=remote_url, - ) - db.add(local_cal) - db.commit() - else: - # Refresh display name and stamp CalDAV metadata if missing. - changed = False - if local_cal.name != display_name: - local_cal.name = display_name - changed = True - if account_id and not local_cal.account_id: - local_cal.account_id = account_id - changed = True - if local_cal.caldav_base_url != remote_url: - local_cal.caldav_base_url = remote_url - changed = True - if changed: - db.commit() - result["calendars"] += 1 - - # Fetch events in window. `date_search` returns CalendarObject - # resources; each may contain one VEVENT (most servers) or - # several (rare). - from icalendar import Calendar as iCal - - seen_uids = set() - # Track events added to the session but not yet committed so - # duplicate UIDs within the same batch are updated, not re-inserted - # (which would violate the UNIQUE constraint on commit). - pending: dict = {} - parse_failed = False - try: - objs = remote_cal.date_search(start=start, end=end, expand=False) - except Exception as e: - result["errors"].append(f"{display_name}: date_search failed ({e})") - continue + calendars = [_open_url_as_calendar(client, url)] + except Exception as e: + result["errors"].append(f"No calendars and URL fallback failed: {e}") + return result # outer finally will call client.close() + + start = datetime.utcnow() - timedelta(days=_LOOKBACK_DAYS) + end = datetime.utcnow() + timedelta(days=_LOOKAHEAD_DAYS) - for obj in objs: + db = SessionLocal() # if this raises, outer finally still calls client.close() + try: + for remote_cal in calendars: + try: + remote_url = str(remote_cal.url) + cal_id = _stable_cal_id(remote_url, owner=owner, account_id=account_id) + display_name = (remote_cal.name or "").strip() or "CalDAV" + + local_cal = db.query(CalendarCal).filter( + CalendarCal.id == cal_id, + CalendarCal.owner == owner, + ).first() + if not local_cal: + local_cal = CalendarCal( + id=cal_id, + owner=owner, + name=display_name, + color="#5b8abf", + source="caldav", + account_id=account_id or None, + caldav_base_url=remote_url, + ) + db.add(local_cal) + db.commit() + else: + # Refresh display name and stamp CalDAV metadata if missing. + changed = False + if local_cal.name != display_name: + local_cal.name = display_name + changed = True + if account_id and not local_cal.account_id: + local_cal.account_id = account_id + changed = True + if local_cal.caldav_base_url != remote_url: + local_cal.caldav_base_url = remote_url + changed = True + if changed: + db.commit() + result["calendars"] += 1 + + # Fetch events in window. `date_search` returns CalendarObject + # resources; each may contain one VEVENT (most servers) or + # several (rare). + from icalendar import Calendar as iCal + + seen_uids = set() + # Track events added to the session but not yet committed so + # duplicate UIDs within the same batch are updated, not re-inserted + # (which would violates the UNIQUE constraint on commit). + pending: dict = {} + parse_failed = False try: - ical = iCal.from_ical(obj.data) + objs = remote_cal.date_search(start=start, end=end, expand=False) except Exception as e: - result["errors"].append(f"{display_name}: parse failed ({e})") - parse_failed = True + result["errors"].append(f"{display_name}: date_search failed ({e})") continue - for comp in ical.walk(): - if comp.name != "VEVENT": + for obj in objs: + try: + ical = iCal.from_ical(obj.data) + except Exception as e: + result["errors"].append(f"{display_name}: parse failed ({e})") + parse_failed = True continue - uid_val = str(comp.get("uid", "")) or str(uuid.uuid4()) - seen_uids.add(uid_val) - dtstart_p = comp.get("dtstart") - if not dtstart_p: - continue - start_dt, all_day = _to_utc_naive(dtstart_p.dt) - - dtend_p = comp.get("dtend") - if dtend_p: - end_dt, _ = _to_utc_naive(dtend_p.dt) - elif all_day: - end_dt = start_dt + timedelta(days=1) - else: - end_dt = start_dt + timedelta(hours=1) - # A synced event with DTEND <= DTSTART (e.g. a single-day - # all-day event whose source wrote DTEND equal to DTSTART) - # would be stored zero-duration and silently dropped by the - # list_events overlap filter. Clamp to a positive span. - end_dt = _ensure_positive_duration(start_dt, end_dt, all_day) - - # is_utc reflects whether the source carried a TZ - # we converted from. All-day = no TZ semantics. - row_is_utc = ( - not all_day - and isinstance(dtstart_p.dt, datetime) - and dtstart_p.dt.tzinfo is not None - ) - - summary = str(comp.get("summary", "")) - description = str(comp.get("description", "")) - location = str(comp.get("location", "")) - rrule = ( - comp.get("rrule").to_ical().decode() - if comp.get("rrule") - else "" - ) + for comp in ical.walk(): + if comp.name != "VEVENT": + continue + uid_val = str(comp.get("uid", "")) or str(uuid.uuid4()) + seen_uids.add(uid_val) - existing = _find_existing_event(db, pending, uid_val, local_cal.id) - if existing: - if existing.caldav_sync_pending in {"create", "update"}: - result["events"] += 1 + dtstart_p = comp.get("dtstart") + if not dtstart_p: continue - existing.calendar_id = local_cal.id - existing.summary = summary - existing.description = description - existing.location = location - existing.dtstart = start_dt - existing.dtend = end_dt - existing.all_day = all_day - existing.is_utc = row_is_utc - existing.rrule = rrule - existing.origin = "caldav" - existing.remote_href = str(getattr(obj, "url", "") or "") or None - existing.remote_etag = _event_etag(obj) or None - existing.caldav_sync_pending = None - else: - new_ev = CalendarEvent( - uid=uid_val, - calendar_id=local_cal.id, - summary=summary, - description=description, - location=location, - dtstart=start_dt, - dtend=end_dt, - all_day=all_day, - is_utc=row_is_utc, - rrule=rrule, - origin="caldav", - remote_href=str(getattr(obj, "url", "") or "") or None, - remote_etag=_event_etag(obj) or None, + start_dt, all_day = _to_utc_naive(dtstart_p.dt) + + dtend_p = comp.get("dtend") + if dtend_p: + end_dt, _ = _to_utc_naive(dtend_p.dt) + elif all_day: + end_dt = start_dt + timedelta(days=1) + else: + end_dt = start_dt + timedelta(hours=1) + # A synced event with DTEND <= DTSTART (e.g. a single-day + # all-day event whose source wrote DTEND equal to DTSTART) + # would be stored zero-duration and silently dropped by the + # list_events overlap filter. Clamp to a positive span. + end_dt = _ensure_positive_duration(start_dt, end_dt, all_day) + + # is_utc reflects whether the source carried a TZ + # we converted from. All-day = no TZ semantics. + row_is_utc = ( + not all_day + and isinstance(dtstart_p.dt, datetime) + and dtstart_p.dt.tzinfo is not None ) - db.add(new_ev) - pending[uid_val] = new_ev - result["events"] += 1 - db.commit() - - # Prune locally-cached CalDAV events that vanished - # upstream (only within our sync window — events outside - # the window aren't in `objs`, so we'd false-delete them). - # Only rows we previously pulled from the server (origin=="caldav") - # are prunable; locally-created events (agent / email triage / a - # UI event whose write-back failed) carry origin NULL and must - # never be deleted just because the server didn't return them. - # Skip the prune on any parse failure: seen_uids is then an - # incomplete view of the server, so pruning against it would - # delete events that still exist upstream but could not be read - # (the empty-seen_uids case wipes the whole window; a partial - # failure deletes just the unreadable rows). - if _should_prune_window(seen_uids, parse_failed): - stale = db.query(CalendarEvent).filter( - CalendarEvent.calendar_id == local_cal.id, - CalendarEvent.origin == "caldav", - CalendarEvent.dtstart >= start, - CalendarEvent.dtstart <= end, - CalendarEvent.remote_href.isnot(None), - CalendarEvent.caldav_sync_pending.is_(None), - ~CalendarEvent.uid.in_(seen_uids) if seen_uids else CalendarEvent.uid.isnot(None), - ).all() - for ev in stale: - db.delete(ev) - result["deleted"] += len(stale) + + summary = str(comp.get("summary", "")) + description = str(comp.get("description", "")) + location = str(comp.get("location", "")) + rrule = ( + comp.get("rrule").to_ical().decode() + if comp.get("rrule") + else "" + ) + + existing = _find_existing_event(db, pending, uid_val, local_cal.id) + if existing: + if existing.caldav_sync_pending in {"create", "update"}: + result["events"] += 1 + continue + existing.calendar_id = local_cal.id + existing.summary = summary + existing.description = description + existing.location = location + existing.dtstart = start_dt + existing.dtend = end_dt + existing.all_day = all_day + existing.is_utc = row_is_utc + existing.rrule = rrule + existing.origin = "caldav" + existing.remote_href = str(getattr(obj, "url", "") or "") or None + existing.remote_etag = _event_etag(obj) or None + existing.caldav_sync_pending = None + else: + new_ev = CalendarEvent( + uid=uid_val, + calendar_id=local_cal.id, + summary=summary, + description=description, + location=location, + dtstart=start_dt, + dtend=end_dt, + all_day=all_day, + is_utc=row_is_utc, + rrule=rrule, + origin="caldav", + remote_href=str(getattr(obj, "url", "") or "") or None, + remote_etag=_event_etag(obj) or None, + ) + db.add(new_ev) + pending[uid_val] = new_ev + result["events"] += 1 db.commit() - except Exception as e: - logger.exception("CalDAV sync failed for one calendar") - result["errors"].append(str(e)[:200]) - db.rollback() - finally: - db.close() - return result + # Prune locally-cached CalDAV events that vanished + # upstream (only within our sync window — events outside + # the window aren't in `objs`, so we'd false-delete them). + # Only rows we previously pulled from the server (origin=="caldav") + # are prunable; locally-created events (agent / email triage / a + # UI event whose write-back failed) carry origin NULL and must + # never be deleted just because the server didn't return them. + # Skip the prune on any parse failure: seen_uids is then an + # incomplete view of the server, so pruning against it would + # delete events that still exist upstream but could not be read + # (the empty-seen_uids case wipes the whole window; a partial + # failure deletes just the unreadable rows). + if _should_prune_window(seen_uids, parse_failed): + stale = db.query(CalendarEvent).filter( + CalendarEvent.calendar_id == local_cal.id, + CalendarEvent.origin == "caldav", + CalendarEvent.dtstart >= start, + CalendarEvent.dtstart <= end, + CalendarEvent.remote_href.isnot(None), + CalendarEvent.caldav_sync_pending.is_(None), + ~CalendarEvent.uid.in_(seen_uids) if seen_uids else CalendarEvent.uid.isnot(None), + ).all() + for ev in stale: + db.delete(ev) + result["deleted"] += len(stale) + db.commit() + except Exception as e: + logger.exception("CalDAV sync failed for one calendar") + result["errors"].append(str(e)[:200]) + db.rollback() + finally: + db.close() # NOT client.close() here anymore + + return result + finally: + client.close() # always called def _event_payload(ev) -> dict: diff --git a/src/caldav_writeback.py b/src/caldav_writeback.py index b1cf288b16..2d5781091d 100644 --- a/src/caldav_writeback.py +++ b/src/caldav_writeback.py @@ -192,11 +192,14 @@ def _writeback_blocking(local_cal_id, ev, delete, url, username, password, # Redirects disabled here too: the write-back path opens its own DAVClient, # so it needs the same SSRF-via-redirect protection as the pull path. client = _build_dav_client(url, username, password) - calendars = _discover_calendars(client) - if not calendars: - return {"ok": False, "error": "no remote calendars discovered"} - return push_event(calendars, local_cal_id, ev, delete=delete, - owner=owner, account_id=account_id) + try: + calendars = _discover_calendars(client) + if not calendars: + return {"ok": False, "error": "no remote calendars discovered"} + return push_event(calendars, local_cal_id, ev, delete=delete, + owner=owner, account_id=account_id) + finally: + client.close() def _persist_writeback_result(owner: str, calendar_id: str, uid: str, result: dict, *, delete: bool) -> None: diff --git a/src/chat_handler.py b/src/chat_handler.py index 9df6f7ce6f..f9b2a21930 100644 --- a/src/chat_handler.py +++ b/src/chat_handler.py @@ -191,6 +191,8 @@ async def preprocess_message( "name": fi.get("name") or fi.get("original_name") or fi["id"], "mime": fi.get("mime", ""), "size": fi.get("size", 0), + "checksum_sha256": fi.get("checksum_sha256") or fi.get("hash"), + "created_at": fi.get("created_at") or fi.get("uploaded_at"), "width": fi.get("width"), "height": fi.get("height"), }) diff --git a/src/config.py b/src/config.py index d5cfa21a7b..fa8b17ce5d 100644 --- a/src/config.py +++ b/src/config.py @@ -27,7 +27,6 @@ class DataConfig(BaseSettings): uploads_dir: Path = Field(default=Path(_DATA_DIR_CONST) / "uploads", description="Directory for uploaded files") sessions_file: Path = Field(default=Path(_DATA_DIR_CONST) / "sessions.json", description="Sessions storage file") memory_file: Path = Field(default=Path(_DATA_DIR_CONST) / "memory.json", description="Memory storage file") - memory_doc: Path = Field(default=Path(_DATA_DIR_CONST) / "memory_doc.md", description="Memory document file") personal_dir: Path = Field(default=Path(_DATA_DIR_CONST) / "personal_docs", description="Personal documents directory") runbook_dir: Path = Field(default=Path(_DATA_DIR_CONST) / "personal_docs" / "runbook", description="Runbook directory") @@ -162,7 +161,6 @@ def set_data_paths(cls, v, info): "uploads_dir": data_dir / "uploads", "sessions_file": data_dir / "sessions.json", "memory_file": data_dir / "memory.json", - "memory_doc": data_dir / "memory_doc.md", "personal_dir": data_dir / "personal_docs", "runbook_dir": data_dir / "personal_docs" / "runbook", "max_upload_size": max_upload_size, diff --git a/src/constants.py b/src/constants.py index eceeb6eb09..f19fb54615 100644 --- a/src/constants.py +++ b/src/constants.py @@ -17,7 +17,6 @@ # re-deriving paths from __file__ or a relative "data" literal. SESSIONS_FILE = os.path.join(DATA_DIR, "sessions.json") MEMORY_FILE = os.path.join(DATA_DIR, "memory.json") -MEMORY_DOC = os.path.join(DATA_DIR, "memory_doc.md") PERSONAL_DIR = os.path.join(DATA_DIR, "personal_docs") RUNBOOK_DIR = os.path.join(PERSONAL_DIR, "runbook") UPLOAD_DIR = os.path.join(DATA_DIR, "uploads") diff --git a/src/copilot.py b/src/copilot.py index 62d2b8ca24..92c9c0a21d 100644 --- a/src/copilot.py +++ b/src/copilot.py @@ -230,7 +230,10 @@ def request_flags(messages) -> tuple: """ msgs = messages or [] last = msgs[-1] if msgs else None - agent = bool(last) and last.get("role") != "user" + # A message element can be a non-dict (clients send `"messages": ["hi"]`); + # the vision loop below already guards each element with isinstance, so do + # the same here rather than call .get() on a bare string. + agent = isinstance(last, dict) and last.get("role") != "user" vision = False for m in msgs: content = m.get("content") if isinstance(m, dict) else None diff --git a/src/document_processor.py b/src/document_processor.py index e96ec999c1..8025e22e03 100644 --- a/src/document_processor.py +++ b/src/document_processor.py @@ -440,7 +440,10 @@ def build_user_content( try: with open(path, "rb") as image_file: encoded_string = base64.b64encode(image_file.read()).decode("utf-8") - image_format = ext[1:] + # Extensionless uploads (e.g. a pasted screenshot) have no ext, + # so fall back to the resolved MIME subtype rather than emitting + # an invalid "data:image/;base64," with an empty subtype. + image_format = ext[1:] or (mime.split("/", 1)[1] if mime.startswith("image/") else "png") content.append({ "type": "image_url", "image_url": {"url": f"data:image/{image_format};base64,{encoded_string}"}, @@ -456,7 +459,7 @@ def build_user_content( try: with open(path, "rb") as audio_file: encoded_string = base64.b64encode(audio_file.read()).decode("utf-8") - audio_format = ext[1:] + audio_format = ext[1:] or (mime.split("/", 1)[1] if mime.startswith("audio/") else "mpeg") content.append({ "type": "audio", "audio": {"url": f"data:audio/{audio_format};base64,{encoded_string}"}, diff --git a/src/llm_core.py b/src/llm_core.py index d8f94dfb74..af1958f16f 100644 --- a/src/llm_core.py +++ b/src/llm_core.py @@ -766,6 +766,36 @@ def apply_kimi_code_headers(headers: Optional[Dict], url: str) -> Dict[str, str] return h +async def apply_kimi_code_headers_async(client, headers: Optional[Dict], url: str) -> Dict[str, str]: + """Pick a Kimi Code User-Agent without blocking the event loop.""" + h = dict(headers or {}) + if not _is_kimi_code_url(url): + return h + base_key = _kimi_code_base_key(url) + cached = _kimi_code_ua_cache.get(base_key) + if cached: + h["User-Agent"] = cached + return h + models_url = base_key.rstrip("/") + "/models" + for ua in KIMI_CODE_USER_AGENTS: + trial = dict(h) + trial["User-Agent"] = ua + try: + r = await client.get(models_url, headers=trial, timeout=8) + except Exception: + continue + if _is_kimi_code_access_denied(r.status_code, r.content): + logger.debug("Kimi Code rejected User-Agent %s (403), trying next", ua) + continue + if r.status_code < 400: + _remember_kimi_code_user_agent(url, ua) + h["User-Agent"] = ua + return h + break + h.setdefault("User-Agent", KIMI_CODE_USER_AGENT) + return h + + def httpx_get_kimi_aware(url: str, headers: Optional[Dict], **kwargs): h = apply_kimi_code_headers(headers, url) if not _is_kimi_code_url(url): @@ -799,7 +829,7 @@ def httpx_post_kimi_aware(url: str, headers: Optional[Dict], **kwargs): async def httpx_post_kimi_aware_async(client, url: str, headers: Optional[Dict], **kwargs): - h = apply_kimi_code_headers(headers, url) + h = await apply_kimi_code_headers_async(client, headers, url) if not _is_kimi_code_url(url): return await client.post(url, headers=h, **kwargs) last = None @@ -2466,9 +2496,9 @@ def _format_routed_content(parts: List[Tuple[str, bool]]) -> List[str]: events.append(_stream_delta_event(part)) return events - h = apply_kimi_code_headers(h, target_url) try: client = _get_http_client() + h = await apply_kimi_code_headers_async(client, h, target_url) async with client.stream('POST', target_url, json=payload, headers=h, timeout=stream_timeout) as r: _clear_host_dead(target_url) if r.status_code != 200: diff --git a/src/mcp_manager.py b/src/mcp_manager.py index 8f4322375d..b71f602ab0 100644 --- a/src/mcp_manager.py +++ b/src/mcp_manager.py @@ -9,7 +9,9 @@ import logging import os import re +import asyncio from typing import Any, Dict, List, Optional, Set, Tuple +from src.database import McpServer, SessionLocal from src.runtime_paths import get_app_root @@ -192,57 +194,65 @@ async def _connect_stdio(self, server_id: str, name: str, command: str, args: Li ) stack = AsyncExitStack() + registered = False + try: transport = await stack.enter_async_context(stdio_client(server_params)) read_stream, write_stream = transport session = await stack.enter_async_context(ClientSession(read_stream, write_stream)) await session.initialize() - - # Discover tools tools_result = await session.list_tools() - except Exception: - await stack.aclose() - raise - tools = [] - for tool in tools_result.tools: - tools.append({ - "name": tool.name, - "description": tool.description or "", - "input_schema": tool.inputSchema if hasattr(tool, 'inputSchema') else {}, - # MCP tool annotations (readOnlyHint / destructiveHint) drive - # plan-mode read-only gating. Absent on many servers, so we - # fall back to a name heuristic in mcp_tool_is_readonly(). - "annotations": getattr(tool, 'annotations', None), - }) - self._sessions[server_id] = session - self._stacks[server_id] = stack - self._tools[server_id] = tools - # Extract identity hints from env vars (e.g. email address, API name) - # so tool descriptions can distinguish between multiple instances of - # the same MCP server (e.g. two email accounts). - identity_hints = [] - for k, v in (env or {}).items(): - k_lower = k.lower() - if any(x in k_lower for x in ['email_address', 'account', 'user', 'username']): - identity_hints.append(v) - identity = ", ".join(identity_hints) if identity_hints else "" + tools = [] + for tool in tools_result.tools: + tools.append({ + "name": tool.name, + "description": tool.description or "", + "input_schema": tool.inputSchema if hasattr(tool, "inputSchema") else {}, + # MCP tool annotations (readOnlyHint / destructiveHint) drive + # plan-mode read-only gating. Absent on many servers, so we + # fall back to a name heuristic in mcp_tool_is_readonly(). + "annotations": getattr(tool, "annotations", None), + }) + + # Extract identity hints from env vars (e.g. email address, API name) + # so tool descriptions can distinguish between multiple instances of + # the same MCP server (e.g. two email accounts). + identity_hints = [] + for k, v in (env or {}).items(): + k_lower = k.lower() + if any(x in k_lower for x in ["email_address", "account", "user", "username"]): + identity_hints.append(v) + identity = ", ".join(identity_hints) if identity_hints else "" + + self._sessions[server_id] = session + self._stacks[server_id] = stack + self._tools[server_id] = tools + self._connections[server_id] = { + "status": "connected", + "name": name, + "transport": "stdio", + "tool_count": len(tools), + "identity": identity, + } - self._connections[server_id] = { - "status": "connected", - "name": name, - "transport": "stdio", - "tool_count": len(tools), - "identity": identity, - } + registered = True + + finally: + if not registered: + await stack.aclose() logger.info(f"MCP server connected: {name} ({server_id}) - {len(tools)} tools via stdio") return True except ImportError: logger.warning("MCP package not installed. Install with: pip install mcp") - self._connections[server_id] = {"status": "error", "error": "mcp package not installed", "name": name} + self._connections[server_id] = { + "status": "error", + "error": "mcp package not installed", + "name": name, + } return False async def _connect_sse(self, server_id: str, name: str, url: str) -> bool: @@ -253,42 +263,46 @@ async def _connect_sse(self, server_id: str, name: str, url: str) -> bool: from contextlib import AsyncExitStack stack = AsyncExitStack() + registered = False + try: transport = await stack.enter_async_context(sse_client(url)) read_stream, write_stream = transport session = await stack.enter_async_context(ClientSession(read_stream, write_stream)) await session.initialize() - - # Discover tools tools_result = await session.list_tools() - except Exception: - await stack.aclose() - raise - tools = [] - for tool in tools_result.tools: - tools.append({ - "name": tool.name, - "description": tool.description or "", - "input_schema": tool.inputSchema if hasattr(tool, 'inputSchema') else {}, - # MCP tool annotations (readOnlyHint / destructiveHint) drive - # plan-mode read-only gating. Absent on many servers, so we - # fall back to a name heuristic in mcp_tool_is_readonly(). - "annotations": getattr(tool, 'annotations', None), - }) - self._sessions[server_id] = session - self._stacks[server_id] = stack - self._tools[server_id] = tools - self._connections[server_id] = { - "status": "connected", - "name": name, - "transport": "sse", - "tool_count": len(tools), - } + tools = [] + for tool in tools_result.tools: + tools.append({ + "name": tool.name, + "description": tool.description or "", + "input_schema": tool.inputSchema if hasattr(tool, 'inputSchema') else {}, + # MCP tool annotations (readOnlyHint / destructiveHint) drive + # plan-mode read-only gating. Absent on many servers, so we + # fall back to a name heuristic in mcp_tool_is_readonly(). + "annotations": getattr(tool, 'annotations', None), + }) + + self._sessions[server_id] = session + self._stacks[server_id] = stack + self._tools[server_id] = tools + self._connections[server_id] = { + "status": "connected", + "name": name, + "transport": "sse", + "tool_count": len(tools), + } - logger.info(f"MCP server connected: {name} ({server_id}) - {len(tools)} tools via SSE") - return True + registered = True + + logger.info(f"MCP server connected: {name} ({server_id}) - {len(tools)} tools via SSE") + return True + + finally: + if not registered: + await stack.aclose() except ImportError: logger.warning("MCP package not installed. Install with: pip install mcp") @@ -340,34 +354,58 @@ def _on_redirect(auth_url): provider = build_provider(server_id, url, on_redirect=_on_redirect) stack = AsyncExitStack() - transport = await stack.enter_async_context(streamablehttp_client(url, auth=provider)) - read_stream, write_stream, _get_session_id = transport - session = await stack.enter_async_context(ClientSession(read_stream, write_stream)) - await session.initialize() - - tools_result = await session.list_tools() - tools = [] - for tool in tools_result.tools: - tools.append({ - "name": tool.name, - "description": tool.description or "", - "input_schema": tool.inputSchema if hasattr(tool, "inputSchema") else {}, - }) + registered = False + try: + transport = await stack.enter_async_context(streamablehttp_client(url, auth=provider)) + read_stream, write_stream, _get_session_id = transport + session = await stack.enter_async_context(ClientSession(read_stream, write_stream)) + await session.initialize() - self._sessions[server_id] = session - self._stacks[server_id] = stack - self._tools[server_id] = tools - self._connections[server_id] = { - "status": "connected", "name": name, "transport": "http", - "tool_count": len(tools), - } - clear_auth_url(server_id) - # Tools changed (this can complete after connect_server already - # returned, via the background OAuth flow), so bump the generation - # to invalidate the tool-prompt cache. - self._generation += 1 - logger.info(f"MCP server connected: {name} ({server_id}) - {len(tools)} tools via http") - return True + tools_result = await session.list_tools() + tools = [] + for tool in tools_result.tools: + tools.append({ + "name": tool.name, + "description": tool.description or "", + "input_schema": tool.inputSchema if hasattr(tool, "inputSchema") else {}, + }) + + self._sessions[server_id] = session + self._stacks[server_id] = stack + self._tools[server_id] = tools + self._connections[server_id] = { + "status": "connected", "name": name, "transport": "http", + "tool_count": len(tools), + } + registered = True + + clear_auth_url(server_id) + # Tools changed (this can complete after connect_server already + # returned, via the background OAuth flow), so bump the generation + # to invalidate the tool-prompt cache. + self._generation += 1 + logger.info(f"MCP server connected: {name} ({server_id}) - {len(tools)} tools via http") + return True + finally: + # _connect_stdio and _connect_sse already guard their stack this way; http + # was the one transport that did not. A failed connect left the stack -- and + # the anyio task group entered inside it -- to be finalized by the garbage + # collector, from a DIFFERENT task than the one that entered it. + # + # anyio's CancelScope.__exit__ then raises "Attempted to exit cancel scope in + # a different task than it was entered in" BEFORE it can remove the host task + # from _tasks, so _deliver_cancellation re-arms call_soon on every iteration + # of the event loop, forever. The loop never sleeps again: + # epoll_wait(timeout=0) at ~149k calls/sec returning zero events, and the + # process pins a CPU core for the rest of its life. + # + # `finally` rather than `except`, deliberately: the failure arrives as + # CancelledError, which is a BaseException, so `except Exception` never sees + # it. (Observed in production: the outer handler below logged nothing at all + # across the entire lifetime of the process, because it could not catch what + # was being raised.) + if not registered: + await stack.aclose() except ImportError: logger.warning("MCP package not installed. Install with: pip install mcp") self._connections[server_id] = {"status": "error", "error": "mcp package not installed", "name": name} @@ -409,17 +447,29 @@ async def disconnect_all(self): for sid in ids: await self.disconnect_server(sid) - async def connect_all_enabled(self): - """Connect to all enabled MCP servers from the database.""" - from src.database import McpServer, SessionLocal + async def connect_all_enabled(self): db = SessionLocal() try: servers = db.query(McpServer).filter(McpServer.is_enabled == True).all() - for srv in servers: - args = json.loads(srv.args) if srv.args else [] - env = json.loads(srv.env) if srv.env else {} - await self.connect_server( + + tasks = [ + asyncio.create_task(self._connect_with_timeout(srv)) + for srv in servers + ] + + await asyncio.gather(*tasks) + finally: + db.close() + + + async def _connect_with_timeout(self, srv): + args = json.loads(srv.args) if srv.args else [] + env = json.loads(srv.env) if srv.env else {} + + try: + await asyncio.wait_for( + self.connect_server( server_id=srv.id, name=srv.name, transport=srv.transport, @@ -427,9 +477,16 @@ async def connect_all_enabled(self): args=args, env=env, url=srv.url, - ) - finally: - db.close() + ), + timeout=20, + ) + except asyncio.TimeoutError: + logger.warning("Timed out connecting to %s", srv.name) + self._connections[srv.id] = { + "status": "timeout", + "error": f"Timed out after 20 seconds", + "name": srv.name, + } async def call_tool(self, qualified_name: str, arguments: Dict) -> Dict: """Call an MCP tool by its qualified name (mcp__{server_id}__{tool_name}). @@ -504,7 +561,7 @@ async def _do_call(self, session, tool_name: str, arguments: Dict) -> Dict: async def _reconnect_builtin(self, server_id: str) -> bool: """Tear down and reconnect a crashed builtin MCP server.""" import sys - from src.builtin_mcp import _BUILTIN_SERVERS + from src.builtin_mcp import _BUILTIN_SERVERS, builtin_python_env if server_id not in _BUILTIN_SERVERS: return False @@ -523,7 +580,7 @@ async def _reconnect_builtin(self, server_id: str) -> bool: transport="stdio", command=sys.executable, args=[script_path], - env={"PYTHONPATH": base_dir}, + env=builtin_python_env(base_dir), ) if ok: logger.info(f"Reconnected builtin MCP server: {name}") diff --git a/src/mcp_oauth.py b/src/mcp_oauth.py index 9f3b2ad4db..27a30383e5 100644 --- a/src/mcp_oauth.py +++ b/src/mcp_oauth.py @@ -96,7 +96,9 @@ def _load(self) -> dict: try: srv = db.query(McpServer).filter(McpServer.id == self.server_id).first() if srv and srv.oauth_tokens: - return json.loads(srv.oauth_tokens) + parsed = json.loads(srv.oauth_tokens) + if isinstance(parsed, dict): + return parsed finally: db.close() return {} @@ -111,6 +113,8 @@ def _update(self, key: str, value: dict) -> None: if srv is None: return data = json.loads(srv.oauth_tokens) if srv.oauth_tokens else {} + if not isinstance(data, dict): + data = {} data[key] = value srv.oauth_tokens = json.dumps(data) db.commit() diff --git a/src/tool_policy.py b/src/tool_policy.py index b70b5c3be8..f0582235a3 100644 --- a/src/tool_policy.py +++ b/src/tool_policy.py @@ -16,6 +16,39 @@ "output they will produce locally." ) +WEB_TOOL_NAMES = frozenset({"web_search", "web_fetch"}) + + +def tool_toggle_enabled(value: object) -> bool: + """Return true only for explicit true-like tool toggle values.""" + + return str(value).lower() == "true" + + +def tool_toggle_explicitly_denied(value: object) -> bool: + """Return true when a caller explicitly supplied a non-true toggle value.""" + + return value is not None and not tool_toggle_enabled(value) + + +def is_web_search_explicitly_denied(allow_web_search: object) -> bool: + """Whether the web-search agent toggle was explicitly set to false.""" + + return tool_toggle_explicitly_denied(allow_web_search) + + +def web_search_enabled_for_turn(allow_web_search: object, use_web: object = None) -> bool: + """Return true only when this request explicitly enables web search. + + Agent mode sends ``allow_web_search``; chat-mode pre-search sends + ``use_web``. If both are present, an explicit ``allow_web_search=false`` + wins so a stale or conflicting intent path cannot re-enable web tools. + """ + + if is_web_search_explicitly_denied(allow_web_search): + return False + return tool_toggle_enabled(allow_web_search) or tool_toggle_enabled(use_web) + _COMMON_TOOL_NAMES = { "api_call", diff --git a/src/tool_utils.py b/src/tool_utils.py index 8255bc0a96..83636cf234 100644 --- a/src/tool_utils.py +++ b/src/tool_utils.py @@ -9,6 +9,7 @@ from src.constants import MAX_OUTPUT_CHARS _mcp_manager = None +_upload_handler = None # --------------------------------------------------------------------------- # MCP Manager singleton @@ -23,6 +24,21 @@ def get_mcp_manager(): """Get the global MCP manager instance.""" return _mcp_manager + +# --------------------------------------------------------------------------- +# Shared upload lifecycle handler +# --------------------------------------------------------------------------- + +def set_upload_handler(handler): + """Register the process's UploadHandler without importing app modules.""" + global _upload_handler + _upload_handler = handler + + +def get_upload_handler(): + """Return the shared UploadHandler used by route and agent writers.""" + return _upload_handler + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- diff --git a/src/tools/calendar.py b/src/tools/calendar.py index d442f2a42a..e6572ba40f 100644 --- a/src/tools/calendar.py +++ b/src/tools/calendar.py @@ -10,6 +10,8 @@ from typing import Dict, Optional from src.tools._common import _parse_tool_args +from src.tool_utils import get_upload_handler +from src.upload_handler import reserve_upload_references logger = logging.getLogger(__name__) @@ -408,11 +410,25 @@ def _create_calendar_reminder(summary: str, location: str, dtstart: datetime, importance = args.get("importance") or "normal" minutes_before = _reminder_minutes(args) + event_description = _event_description(args, minutes_before) + event_location = args.get("location", "") or "" + missing_id = reserve_upload_references( + get_upload_handler(), + owner, + event_description, + event_location, + ) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } + uid = str(_uuid.uuid4()) ev = CalendarEvent( uid=uid, calendar_id=cal.id, summary=summary, - description=_event_description(args, minutes_before), - location=args.get("location", "") or "", + description=event_description, + location=event_location, dtstart=dtstart, dtend=dtend, all_day=all_day, is_utc=dtstart_is_utc and not all_day, rrule=args.get("rrule", "") or "", @@ -465,6 +481,17 @@ def _create_calendar_reminder(summary: str, location: str, dtstart: datetime, ev = _event_query().filter(CalendarEvent.uid == base_uid).first() if not ev: return {"error": f"Event {uid} not found", "exit_code": 1} + missing_id = reserve_upload_references( + get_upload_handler(), + owner, + args.get("description"), + args.get("location"), + ) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } if args.get("summary") is not None: ev.summary = args["summary"] if args.get("description") is not None: diff --git a/src/tools/notes.py b/src/tools/notes.py index d24be55d1d..ba980c158e 100644 --- a/src/tools/notes.py +++ b/src/tools/notes.py @@ -10,6 +10,8 @@ from typing import Dict, Optional from src.tools._common import _parse_tool_args +from src.tool_utils import get_upload_handler +from src.upload_handler import reserve_upload_references logger = logging.getLogger(__name__) @@ -198,6 +200,18 @@ def _format_note_list(notes) -> str: "duplicate": True, "exit_code": 0, } + missing_id = reserve_upload_references( + get_upload_handler(), + owner, + content_raw, + args.get("color"), + items_json, + ) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } note = Note( id=str(_uuid.uuid4()), owner=owner, @@ -235,6 +249,19 @@ def _format_note_list(notes) -> str: return {"error": f"Note '{note_id}' not found", "exit_code": 1} if not _note_visible_to_owner(note, owner): return {"error": "Note not found", "exit_code": 1} + missing_id = reserve_upload_references( + get_upload_handler(), + owner, + args.get("content"), + args.get("color"), + args.get("checklist_items"), + args.get("items"), + ) + if missing_id: + return { + "error": f"Referenced upload is no longer available: {missing_id}", + "exit_code": 1, + } for field in ("title", "content", "note_type", "color", "label"): if field in args and args[field] is not None: setattr(note, field, args[field]) diff --git a/src/tools/system.py b/src/tools/system.py index 3fedac6c37..a901992c26 100644 --- a/src/tools/system.py +++ b/src/tools/system.py @@ -356,7 +356,11 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + # Strict ownership: the old `task.owner and task.owner != owner` + # skipped the check on an owner-less task (created in no-login mode + # or before the legacy-owner sweep), letting any authenticated user + # reach it. `list` already scopes to an exact owner match. + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} changed = [] @@ -402,7 +406,7 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} name = task.name db.delete(task) @@ -416,7 +420,7 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} if action == "pause": @@ -437,7 +441,7 @@ async def do_manage_tasks(content: str, owner: Optional[str] = None) -> Dict: task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first() if not task: return {"error": f"Task {task_id} not found", "exit_code": 1} - if owner and task.owner and task.owner != owner: + if owner and task.owner != owner: return {"error": "Access denied", "exit_code": 1} from src.event_bus import get_task_scheduler diff --git a/src/upload_handler.py b/src/upload_handler.py index 1f24c62634..ce0b4b1293 100644 --- a/src/upload_handler.py +++ b/src/upload_handler.py @@ -35,11 +35,34 @@ def secure_filename(filename: str) -> str: logger = logging.getLogger(__name__) + +class UploadCleanupSafetyError(RuntimeError): + """Raised when cleanup cannot prove that destructive work is safe.""" + # The extension is optional: save_upload builds the id as `{uuid.hex}{ext}`, # and a file with no extension (Dockerfile, README, ...) yields a bare 32-hex # id. Requiring `.ext` made those ids fail validation, so the stored file # could never be resolved or downloaded again. UPLOAD_ID_RE = re.compile(r"^[0-9a-fA-F]{32}(?:\.[A-Za-z0-9]+)?$") +UPLOAD_ID_TOKEN_RE = re.compile( + r"(?\[\](){},;!?:&#]|\.(?![A-Za-z0-9]))" +) +PDF_SOURCE_UPLOAD_RE = re.compile( + r"", + re.IGNORECASE, +) +ATTACHMENT_REFERENCE_LINE_RE = re.compile( + r"\[Attachment:[^\]\r\n]*\|\s*id=" + r"([0-9a-fA-F]{32}(?:\.[A-Za-z0-9]+)?)" + r"(?:\s*\||\s*\])", + re.IGNORECASE, +) def is_valid_upload_id(upload_id: str) -> bool: @@ -47,6 +70,110 @@ def is_valid_upload_id(upload_id: str) -> bool: return UPLOAD_ID_RE.fullmatch(upload_id or "") is not None +def extract_upload_ids(value: Any) -> set[str]: + """Return canonical upload IDs embedded in a persisted URL/text value.""" + if not isinstance(value, str) or not value: + return set() + return set(UPLOAD_ID_TOKEN_RE.findall(value)) + + +def extract_internal_upload_ids(value: Any) -> set[str]: + """Return IDs from explicit internal upload references only. + + Cleanup intentionally uses :func:`extract_upload_ids` conservatively, but + write-time reservation must not treat an arbitrary 32-hex checksum in note + or calendar text as an upload reference. Nested JSON-like values are + supported because note checklist items are persisted as structured data. + """ + if isinstance(value, dict): + found: set[str] = set() + for nested in value.values(): + found.update(extract_internal_upload_ids(nested)) + return found + if isinstance(value, (list, tuple, set)): + found: set[str] = set() + for nested in value: + found.update(extract_internal_upload_ids(nested)) + return found + if not isinstance(value, str) or not value: + return set() + return ( + set(INTERNAL_UPLOAD_URL_RE.findall(value)) + | set(PDF_SOURCE_UPLOAD_RE.findall(value)) + | set(ATTACHMENT_REFERENCE_LINE_RE.findall(value)) + ) + + +def reserve_upload_references( + upload_handler: Any, + owner: Optional[str], + *values: Any, +) -> Optional[str]: + """Reserve upload IDs in values before a caller persists references. + + Returns the first ID that cannot be owner-checked/reserved, otherwise + ``None``. A missing handler is treated as no-op for backward-compatible + route factories; production wires the shared UploadHandler instance. + """ + if upload_handler is None: + return None + upload_ids: set[str] = set() + for value in values: + upload_ids.update(extract_internal_upload_ids(value)) + return reserve_upload_ids(upload_handler, owner, upload_ids) + + +def reserve_upload_ids( + upload_handler: Any, + owner: Optional[str], + upload_ids: Any, +) -> Optional[str]: + """Owner-reserve canonical IDs from a trusted structured reference field.""" + if upload_handler is None: + return None + canonical_ids = { + str(upload_id).strip() + for upload_id in (upload_ids or []) + if is_valid_upload_id(str(upload_id).strip()) + } + for upload_id in sorted(canonical_ids): + try: + resolved = upload_handler.reserve_upload( + upload_id, + owner=owner, + allow_admin=False, + ) + except Exception: + resolved = None + if not resolved: + return upload_id + return None + + +def reserve_message_upload_references( + upload_handler: Any, + owner: Optional[str], + content: Any, + metadata: Any = None, +) -> Optional[str]: + """Reserve explicit chat references, including structured attachment IDs.""" + upload_ids = extract_internal_upload_ids(content) + if metadata not in (None, ""): + if isinstance(metadata, str): + metadata = json.loads(metadata) + if not isinstance(metadata, dict): + raise ValueError("message metadata must be a JSON object") + upload_ids.update(extract_internal_upload_ids(metadata)) + from src.attachment_refs import attachment_refs_from_metadata + + upload_ids.update( + str(ref.get("attachment_id") or "").strip() + for ref in attachment_refs_from_metadata(metadata) + if ref.get("attachment_id") + ) + return reserve_upload_ids(upload_handler, owner, upload_ids) + + def _build_upload_id(safe_filename: str) -> str: """Build a unique upload id whose extension matches UPLOAD_ID_RE. @@ -249,43 +376,295 @@ def is_safe_file_type(self, content_type: str, filename: str) -> bool: return True - def cleanup_old_uploads(self): - """Remove uploaded files older than CLEANUP_DAYS days.""" + @staticmethod + def _parse_upload_timestamp(value: Any) -> Optional[datetime]: + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + if parsed.tzinfo is not None: + parsed = parsed.astimezone().replace(tzinfo=None) + return parsed + except (TypeError, ValueError): + return None + + @classmethod + def _upload_metadata_is_recent(cls, info: Dict[str, Any], cutoff_date: datetime) -> bool: + """Return True when upload metadata records activity inside retention.""" + for field in ("last_accessed", "created_at", "uploaded_at"): + parsed = cls._parse_upload_timestamp(info.get(field)) + if parsed is None: + continue + if parsed >= cutoff_date: + return True + return False + + @classmethod + def _upload_index_keys_for_file( + cls, + upload_index: Dict[str, Any], + upload_id: str, + file_path: str, + ) -> list[str]: + """Find a coherent set of index rows for one physical upload. + + Every related row must agree on ID, canonical path, owner, and a + non-empty checksum. Each row must also contain the complete lifecycle + timestamps written for new uploads. Ambiguous or incomplete index + state cannot authorize destructive cleanup. + """ + target_path = os.path.normcase(os.path.realpath(file_path)) + matches: list[str] = [] + owners: set[str] = set() + checksums: set[str] = set() + for key, info in upload_index.items(): + if not isinstance(info, dict): + continue + stored_path = info.get("path") + stored_real_path = ( + os.path.normcase(os.path.realpath(stored_path)) + if isinstance(stored_path, str) and stored_path + else None + ) + same_id = info.get("id") == upload_id + same_path = stored_real_path == target_path + if not same_id and not same_path: + continue + if not same_id or not same_path: + logger.warning( + "Skipping ambiguous cleanup candidate %s: related row has id=%r path=%r", + file_path, + info.get("id"), + stored_path, + ) + return [] + + owner = info.get("owner") + if not isinstance(owner, str) or not owner.strip(): + logger.warning( + "Skipping incomplete cleanup candidate %s: matching row has no owner", + file_path, + ) + return [] + + row_checksums = { + str(info.get(field)).strip().lower() + for field in ("hash", "checksum_sha256") + if info.get(field) is not None and str(info.get(field)).strip() + } + if not row_checksums: + logger.warning( + "Skipping incomplete cleanup candidate %s: matching row has no checksum", + file_path, + ) + return [] + if len(row_checksums) != 1: + logger.warning( + "Skipping ambiguous cleanup candidate %s: matching row has conflicting checksums", + file_path, + ) + return [] + + lifecycle_fields = ("uploaded_at", "created_at", "last_accessed") + if any( + cls._parse_upload_timestamp(info.get(field)) is None + for field in lifecycle_fields + ): + logger.warning( + "Skipping incomplete cleanup candidate %s: matching row lacks lifecycle timestamps", + file_path, + ) + return [] + + matches.append(key) + owners.add(owner) + checksums.update(row_checksums) + + if len(owners) > 1 or len(checksums) > 1: + logger.warning( + "Skipping ambiguous cleanup candidate %s: matching rows disagree on owner or checksum", + file_path, + ) + return [] + return matches + + def cleanup_old_uploads( + self, + referenced_upload_ids: Optional[set[str]] = None, + referenced_upload_hashes: Optional[set[str]] = None, + ): + """Remove expired uploads proven unreferenced by a complete snapshot. + + ``None`` means reference discovery was not completed, so cleanup fails + closed and removes nothing. The admin route supplies both sets after + scanning persisted chats, documents, and gallery records. + """ + if referenced_upload_ids is None or referenced_upload_hashes is None: + logger.warning("Upload cleanup skipped: persisted reference snapshot unavailable") + return 0 + try: - cutoff_date = datetime.now() - timedelta(days=self.cleanup_days) + cleanup_started_at = datetime.now() + cutoff_date = cleanup_started_at - timedelta(days=self.cleanup_days) cleaned_count = 0 - - for root, dirs, files in os.walk(self.upload_dir): - if root == self.upload_dir: - continue - - path_parts = root.split(os.sep) - if len(path_parts) >= 4: + + referenced_ids = {str(value) for value in referenced_upload_ids} + referenced_hashes = {str(value) for value in referenced_upload_hashes} + uploads_db_path = os.path.join(self.upload_dir, "uploads.json") + + # Keep index mutation and file removal serialized with upload writes. + # Each row removal is atomically persisted before the bytes are + # deleted; if deletion fails, the previous index is restored. + with self._index_lock: + current_index = dict(self._load_upload_index(fail_on_error=True)) + + for root, dirs, files in os.walk(self.upload_dir, followlinks=False): + is_junction = getattr(os.path, "isjunction", lambda _path: False) + dirs[:] = [ + directory + for directory in dirs + if not os.path.islink(os.path.join(root, directory)) + and not is_junction(os.path.join(root, directory)) + ] + if root == self.upload_dir: + continue + if not self._inside_upload_dir(root): + dirs[:] = [] + continue + + path_parts = root.split(os.sep) + if len(path_parts) < 4: + continue try: dir_date = datetime(int(path_parts[-3]), int(path_parts[-2]), int(path_parts[-1])) - if dir_date < cutoff_date: - for file in files: - file_path = os.path.join(root, file) - try: - os.remove(file_path) - cleaned_count += 1 - logger.info(f"Cleaned up old upload: {file_path}") - except Exception as e: - logger.warning(f"Failed to remove {file_path}: {e}") - - try: - os.rmdir(root) - logger.info(f"Removed empty upload directory: {root}") - except Exception as e: - logger.warning(f"Failed to remove directory {root}: {e}") except (ValueError, IndexError): continue - + if dir_date >= cutoff_date: + continue + + for file in files: + # Reference discovery only understands canonical upload + # IDs; unknown files fail closed instead of being swept. + if not self.validate_upload_id(file): + continue + + file_path = os.path.join(root, file) + if not self._inside_upload_dir(file_path): + logger.warning( + "Skipping cleanup candidate outside upload directory: %s", + file_path, + ) + continue + matching_keys = self._upload_index_keys_for_file( + current_index, + file, + file_path, + ) + matching_rows = [ + current_index[key] + for key in matching_keys + if isinstance(current_index.get(key), dict) + ] + + # Files without authoritative live index rows are not + # eligible for destructive cleanup. Reference hashes, + # recency, and ownership cannot be proven for them. + if not matching_rows: + continue + + is_referenced = file in referenced_ids or any( + str(info.get("id") or "") in referenced_ids + or str(info.get("hash") or "") in referenced_hashes + or str(info.get("checksum_sha256") or "") in referenced_hashes + for info in matching_rows + ) + metadata_is_recent = any( + self._upload_metadata_is_recent(info, cutoff_date) + for info in matching_rows + ) + if is_referenced or metadata_is_recent: + continue + + reduced_index = { + key: value + for key, value in current_index.items() + if key not in matching_keys + } + if matching_keys: + try: + self._atomic_write_json( + uploads_db_path, + reduced_index, + sync_backup=True, + ) + except Exception as e: + try: + self._atomic_write_json( + uploads_db_path, + current_index, + sync_backup=True, + ) + except Exception: + logger.exception( + "Failed to restore upload indexes after reconciliation failed for %s", + file_path, + ) + raise UploadCleanupSafetyError( + "upload index rollback failed before file removal" + ) from e + logger.warning( + "Failed to reconcile upload index before removing %s: %s", + file_path, + e, + ) + continue + + try: + os.remove(file_path) + except FileNotFoundError: + # The bytes are already absent. Keep the reduced + # lifecycle index instead of recreating a stale row. + current_index = reduced_index + logger.info( + "Reconciled missing expired upload from index: %s", + file_path, + ) + continue + except Exception as e: + if matching_keys: + try: + self._atomic_write_json( + uploads_db_path, + current_index, + sync_backup=True, + ) + except Exception: + logger.exception( + "Failed to restore upload index after removal failed for %s", + file_path, + ) + raise UploadCleanupSafetyError( + "upload index rollback failed after file removal was refused" + ) from e + logger.warning(f"Failed to remove {file_path}: {e}") + continue + + current_index = reduced_index + cleaned_count += 1 + logger.info(f"Cleaned up old unreferenced upload: {file_path}") + + try: + if not os.listdir(root): + os.rmdir(root) + logger.info(f"Removed empty upload directory: {root}") + except Exception as e: + logger.warning(f"Failed to inspect/remove directory {root}: {e}") + logger.info(f"Upload cleanup completed: {cleaned_count} files removed") return cleaned_count except Exception as e: logger.error(f"Upload cleanup failed: {e}") - return 0 + raise UploadCleanupSafetyError("upload cleanup safety checks failed") from e def validate_upload_id(self, upload_id: str) -> bool: """Validate that the upload ID matches the expected pattern.""" @@ -293,71 +672,103 @@ def validate_upload_id(self, upload_id: str) -> bool: def _inside_upload_dir(self, path: str) -> bool: """Check if path is inside the upload directory.""" - base = os.path.realpath(self.upload_dir) - p = os.path.realpath(path) + base = os.path.normcase(os.path.realpath(self.upload_dir)) + p = os.path.normcase(os.path.realpath(path)) try: return os.path.commonpath([base, p]) == base except Exception: return False - def _atomic_write_json(self, path: str, data: dict) -> None: + def _atomic_write_json( + self, + path: str, + data: dict, + *, + sync_backup: bool = False, + ) -> None: """Write `data` to `path` atomically: write to a temp file in the same directory, then `os.replace` onto the target. The kernel guarantees `os.replace` is atomic on POSIX, so a reader either sees the old contents or the new contents, never a half-written - file. Also keeps a `.bak` sibling of the previous good state. + file. Normally `.bak` retains the previous good state. Destructive + lifecycle transitions use ``sync_backup=True`` so recovery cannot + resurrect metadata for bytes that were deliberately removed. """ directory = os.path.dirname(path) or "." - fd, tmp = tempfile.mkstemp(prefix=".uploads-", suffix=".tmp", dir=directory) - try: - with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2) - f.flush() - os.fsync(f.fileno()) - if os.path.exists(path): - bak = path + ".bak" + + def _replace_json(target: str) -> None: + fd, tmp = tempfile.mkstemp( + prefix=".uploads-", + suffix=".tmp", + dir=directory, + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp, target) + except Exception: try: - shutil.copy2(path, bak) + os.unlink(tmp) except OSError: pass - os.replace(tmp, path) - # Update cache if this is the main index - if path.endswith("uploads.json"): - self._index_cache = data - try: - self._index_mtime = os.path.getmtime(path) - except OSError: - self._index_mtime = time.time() - except Exception: + raise + + if sync_backup: + _replace_json(path + ".bak") + elif os.path.exists(path): try: - os.unlink(tmp) + shutil.copy2(path, path + ".bak") except OSError: pass - raise - def _load_upload_index(self) -> Dict[str, Any]: + _replace_json(path) + # Update cache if this is the main index + if path.endswith("uploads.json"): + self._index_cache = data + try: + self._index_mtime = os.path.getmtime(path) + except OSError: + self._index_mtime = time.time() + + def _load_upload_index(self, *, fail_on_error: bool = False) -> Dict[str, Any]: """Load the upload index from disk/cache. Uses mtime-based validation - to avoid redundant parsing on hot paths. + to avoid redundant parsing on hot paths. When ``fail_on_error`` is + true, a missing, malformed, or unreadable live index raises so + destructive callers cannot mistake corruption for an empty store. """ uploads_db_path = os.path.join(self.upload_dir, "uploads.json") - if not os.path.exists(uploads_db_path): + candidates = (uploads_db_path, uploads_db_path + ".bak") + if fail_on_error: + # A backup is intentionally the previous snapshot. It is useful for + # non-destructive reads, but cannot authorize deletion when the live + # index is missing or corrupt. + if not os.path.exists(uploads_db_path): + raise ValueError("live uploads database is missing") + existing_candidates = [uploads_db_path] + else: + existing_candidates = [path for path in candidates if os.path.exists(path)] + if not existing_candidates: self._index_cache = {} self._index_mtime = 0.0 return {} # Check cache validity try: - mtime = os.path.getmtime(uploads_db_path) - if self._index_cache is not None and mtime <= self._index_mtime: + mtime = max(os.path.getmtime(path) for path in existing_candidates) + if ( + not fail_on_error + and self._index_cache is not None + and mtime <= self._index_mtime + ): return self._index_cache except OSError: mtime = 0.0 # Try the live file first, fall back to the .bak sibling if the # live file is truncated/corrupted. - for candidate in (uploads_db_path, uploads_db_path + ".bak"): - if not os.path.exists(candidate): - continue + for candidate in existing_candidates: try: with open(candidate, "r", encoding="utf-8") as f: data = json.load(f) @@ -369,6 +780,8 @@ def _load_upload_index(self) -> Dict[str, Any]: logger.warning(f"Failed to read uploads database ({candidate}): {e}") continue + if fail_on_error: + raise ValueError("live uploads database is unreadable") self._index_cache = {} return {} @@ -381,6 +794,144 @@ def get_upload_info(self, upload_id: str) -> Optional[Dict[str, Any]]: return dict(info) return None + def reserve_upload( + self, + upload_id: str, + *, + owner: Optional[str], + auth_manager: Any = None, + allow_admin: bool = False, + ) -> Optional[Dict[str, Any]]: + """Owner-check and reserve an indexed upload against cleanup. + + The live index lookup, ownership/path validation, and access touch all + occur under the cleanup lock. A durable-reference writer must not + commit when this returns ``None``. + """ + if not self.validate_upload_id(upload_id): + return None + + auth_configured = bool(auth_manager and getattr(auth_manager, "is_configured", False)) + if auth_configured and not owner: + return None + + uploads_db_path = os.path.join(self.upload_dir, "uploads.json") + with self._index_lock: + try: + current = dict(self._load_upload_index(fail_on_error=True)) + except Exception: + logger.warning("Cannot reserve upload %s without a valid live index", upload_id) + return None + matching_keys = [ + key + for key, info in current.items() + if isinstance(info, dict) and info.get("id") == upload_id + ] + if not matching_keys: + return None + + matching_rows = [dict(current[key]) for key in matching_keys] + row_owners = { + str(row.get("owner")) if row.get("owner") is not None else None + for row in matching_rows + } + row_hashes = { + str(row.get("hash") or row.get("checksum_sha256")) + for row in matching_rows + if row.get("hash") or row.get("checksum_sha256") + } + if len(row_owners) != 1 or len(row_hashes) > 1: + logger.warning( + "Cannot reserve ambiguous upload index rows for %s", + upload_id, + ) + return None + + is_admin = False + if allow_admin and owner and auth_manager and hasattr(auth_manager, "is_admin"): + try: + is_admin = bool(auth_manager.is_admin(owner)) + except Exception: + is_admin = False + + now = datetime.now() + current_info = matching_rows[0] + if owner and not is_admin and current_info.get("owner") != owner: + return None + if not owner and current_info.get("owner") is not None: + return None + + existing_paths: set[str] = set() + for row in matching_rows: + stored_path = row.get("path") + if not stored_path: + continue + if not self._inside_upload_dir(stored_path): + logger.warning( + "Cannot reserve upload %s with an out-of-root index path", + upload_id, + ) + return None + if os.path.isfile(stored_path): + if os.path.basename(stored_path) != upload_id: + return None + existing_paths.add(os.path.normcase(os.path.realpath(stored_path))) + if len(existing_paths) > 1: + logger.warning("Cannot reserve upload %s with multiple indexed paths", upload_id) + return None + path = next(iter(existing_paths), None) or self._find_upload_path(upload_id) + if not path or not os.path.isfile(path) or not self._inside_upload_dir(path): + return None + + last_accessed = self._parse_upload_timestamp(current_info.get("last_accessed")) + path_changed = current_info.get("path") != path + needs_write = ( + path_changed + or last_accessed is None + or last_accessed < now - timedelta(minutes=5) + ) + if needs_write: + accessed_at = now.isoformat() + updated_index = dict(current) + for key in matching_keys: + updated = dict(updated_index[key]) + updated["path"] = path + updated["last_accessed"] = accessed_at + updated_index[key] = updated + try: + self._atomic_write_json( + uploads_db_path, + updated_index, + sync_backup=True, + ) + except Exception: + try: + self._atomic_write_json( + uploads_db_path, + current, + sync_backup=True, + ) + except Exception: + logger.exception( + "Failed to restore upload indexes after reservation failed for %s", + upload_id, + ) + logger.exception("Failed to reserve upload %s against cleanup", upload_id) + return None + current_info = dict(updated_index[matching_keys[0]]) + + resolved = dict(current_info) + resolved.setdefault("id", upload_id) + resolved["path"] = path + resolved.setdefault("name", os.path.basename(path)) + resolved.setdefault("original_name", resolved["name"]) + resolved.setdefault("mime", mimetypes.guess_type(path)[0] or "application/octet-stream") + if resolved.get("hash") and not resolved.get("checksum_sha256"): + resolved["checksum_sha256"] = resolved["hash"] + if resolved.get("uploaded_at") and not resolved.get("created_at"): + resolved["created_at"] = resolved["uploaded_at"] + return resolved + def _renamed_upload_index_key(self, key: str, info: Dict[str, Any], old_owner: str, new_owner: str) -> str: """Return the storage key to use after renaming an owned upload row. @@ -475,16 +1026,32 @@ def _find_upload_path(self, upload_id: str) -> Optional[str]: if not self.validate_upload_id(upload_id): return None + candidates: list[str] = [] direct = os.path.join(self.upload_dir, upload_id) - if os.path.exists(direct) and self._inside_upload_dir(direct): - return direct + if os.path.isfile(direct) and self._inside_upload_dir(direct): + candidates.append(os.path.realpath(direct)) - for root, _dirs, files in os.walk(self.upload_dir, followlinks=False): + for root, dirs, files in os.walk(self.upload_dir, followlinks=False): + is_junction = getattr(os.path, "isjunction", lambda _path: False) + dirs[:] = [ + directory + for directory in dirs + if not os.path.islink(os.path.join(root, directory)) + and not is_junction(os.path.join(root, directory)) + ] if upload_id in files: path = os.path.join(root, upload_id) - if self._inside_upload_dir(path): - return path - return None + if os.path.isfile(path) and self._inside_upload_dir(path): + real_path = os.path.realpath(path) + if real_path not in candidates: + candidates.append(real_path) + if len(candidates) > 1: + logger.warning( + "Upload ID %s resolves to multiple physical files", + upload_id, + ) + return None + return candidates[0] if candidates else None def resolve_upload( self, @@ -493,52 +1060,20 @@ def resolve_upload( auth_manager: Any = None, allow_admin: bool = True, ) -> Optional[Dict[str, Any]]: - """Resolve an upload ID to metadata only if the caller may read it. + """Resolve and reserve an upload only if the caller may read it. This is the owner-aware lookup used by internal processors. Public download routes already perform owner checks; chat/document paths must - do the same before reading file bytes server-side. + do the same before reading file bytes server-side. Reservation shares + cleanup's lifecycle lock and prevents a newly persisted reference from + racing final deletion. """ - if not self.validate_upload_id(upload_id): - logger.warning(f"Invalid upload ID format: {upload_id}") - return None - - auth_configured = bool(auth_manager and getattr(auth_manager, "is_configured", False)) - if auth_configured and not owner: - return None - - info = self.get_upload_info(upload_id) or {} - is_admin = False - if allow_admin and owner and auth_manager and hasattr(auth_manager, "is_admin"): - try: - is_admin = bool(auth_manager.is_admin(owner)) - except Exception: - is_admin = False - - if owner and not is_admin: - if info.get("owner") != owner: - logger.warning("Upload %s denied for owner %s", upload_id, owner) - return None - if not owner and info.get("owner") is not None: - logger.warning("Upload %s denied without an authenticated owner", upload_id) - return None - - path = info.get("path") - if not path or not os.path.exists(path) or not self._inside_upload_dir(path): - path = self._find_upload_path(upload_id) - if not path: - return None - if not self._inside_upload_dir(path): - logger.warning(f"Upload path outside upload directory: {path}") - return None - - resolved = dict(info) - resolved.setdefault("id", upload_id) - resolved["path"] = path - resolved.setdefault("name", os.path.basename(path)) - resolved.setdefault("original_name", resolved["name"]) - resolved.setdefault("mime", mimetypes.guess_type(path)[0] or "application/octet-stream") - return resolved + return self.reserve_upload( + upload_id, + owner=owner, + auth_manager=auth_manager, + allow_admin=allow_admin, + ) def cleanup_rate_limits(self): """Remove stale entries from upload_rate_log.""" @@ -706,6 +1241,9 @@ def save_upload(self, u: UploadFile, client_ip: str, owner: str = None) -> dict: # fresh-insert path below; release the lock first. raise LookupError("upload entry vanished mid-dedupe") existing_file["last_accessed"] = datetime.now().isoformat() + existing_file.setdefault("checksum_sha256", file_hash) + if existing_file.get("uploaded_at"): + existing_file.setdefault("created_at", existing_file["uploaded_at"]) current[live_key] = existing_file self._atomic_write_json(uploads_db_path, current) except LookupError: @@ -721,7 +1259,9 @@ def save_upload(self, u: UploadFile, client_ip: str, owner: str = None) -> dict: "size": existing_file["size"], "name": existing_file["original_name"], "hash": file_hash, + "checksum_sha256": existing_file.get("checksum_sha256") or file_hash, "uploaded_at": existing_file["uploaded_at"], + "created_at": existing_file.get("created_at") or existing_file["uploaded_at"], "owner": existing_file.get("owner"), "width": existing_file.get("width"), "height": existing_file.get("height"), @@ -744,6 +1284,7 @@ def save_upload(self, u: UploadFile, client_ip: str, owner: str = None) -> dict: raise HTTPException(status_code=500, detail=f"Failed to save file: {str(e)}") # Create file metadata + created_at = datetime.now().isoformat() file_metadata = { "id": file_id, "path": file_path, @@ -751,9 +1292,11 @@ def save_upload(self, u: UploadFile, client_ip: str, owner: str = None) -> dict: "size": file_size, "name": safe_filename, "hash": file_hash, + "checksum_sha256": file_hash, "original_name": original_filename, - "uploaded_at": datetime.now().isoformat(), - "last_accessed": datetime.now().isoformat(), + "uploaded_at": created_at, + "created_at": created_at, + "last_accessed": created_at, "client_ip": client_ip, "owner": owner, } diff --git a/static/js/MODULE_SUMMARY.md b/static/js/MODULE_SUMMARY.md index 0e847423fa..df5b0cb33c 100644 --- a/static/js/MODULE_SUMMARY.md +++ b/static/js/MODULE_SUMMARY.md @@ -1,124 +1,228 @@ -# Module Organization Summary +# Frontend Module Organization Summary -## Purpose -This document describes what each JavaScript module is responsible for. +> **Scope:** This document describes the architecture of the Odysseus no-build +> frontend. The app is a collection of native ES6 modules loaded from +> `static/`. The authoritative source is the current `static/js/` tree and the +> top-level orchestrator `static/app.js`. -> **Note:** This file is a partial, historical overview — not a complete authoritative -> inventory. The authoritative module set is the current `static/js/` tree plus the -> scripts loaded by `static/index.html`. As of this writing that tree holds **65 `.js` -> files** across **8 subdirectories** (`calendar/`, `color/`, `compare/`, `editor/`, -> `emailLibrary/`, `markdown/`, `research/`, `util/`), and `static/index.html` loads -> **35** `/static…` script tags. The catalog below covers only the original core -> modules and is not kept in sync with every module. +--- + +## 1. Top-level Application Orchestrator + +### `static/app.js` +*Main application entry point.* + +- Imports all feature modules. +- Exposes a few modules on `window` for legacy inter-module reachability + (`themeModule`, `sessionModule`, `uiModule`, `adminModule`, `cookbookModule`). +- Patches `fetch` so any `401` redirects the user to `/login`. +- Fetches the default chat configuration and handles deep-link route openers + (`/notes`, `/calendar`, `/email`, `/memory`, `/gallery`, `/cookbook`, `/library`, `/tasks`). +- Wires global event listeners: chat-history scrolling, popups, Escape handling, + drag-and-drop/paste attachment handling, transcription export, sidebar toggles, + rail/tool buttons, and session sorting. +- Loads auth status and applies per-user privilege restrictions. + +### `static/index.html` +*SPA shell.* Loads `app.js` as a module, includes the theme-aware inline script, +and defines the DOM skeleton that the modules populate (chat history, composer, +sidebar, icon rail, modals). + +--- + +## 2. Core Foundation Modules + +These are imported first and used across most features. + +| Module | Primary Exports | Responsibility | +|---|---|---| +| **`ui.js`** | `showToast`, `showError`, `el`, `copyToClipboard`, `scrollHistory`, `setAutoScroll`, `autoResize`, `debounce`, `esc` | Shared UI helpers, toast notifications, scroll behavior, element accessor, text escaping. | +| **`storage.js`** | `default` storage wrapper | LocalStorage helpers and toggle state persistence. | +| **`markdown.js`** | `mdToHtml`, `processWithThinking`, `squashOutsideCode`, `normalizeThinkingMarkup`, `extractThinkingBlocks`, `hasUnclosedThinkTag`, `startsWithReasoningPrefix` | Markdown→HTML, thinking/reasoning block parsing, code-block normalization. | +| **`spinner.js`** | `create`, `createWhirlpool` | Loading/spinner factories for streaming and tool cards. | +| **`keyboard-shortcuts.js`** | `initKeyboardShortcuts` | Global keyboard shortcut wiring. | +| **`sidebar-layout.js`** | `initSidebarLayout`, `syncRailSide` | Wide sidebar ↔ icon-rail layout behavior. | +| **`section-management.js`** | `initSectionCollapse`, `initSectionDrag` | Collapsible/draggable sidebar sections. | +| **`modalManager.js`** | side-effect import | Unified minimize/restore behavior for floating tool modals. | +| **`tileManager.js`** | side-effect import | Desktop window tiling and snap-to-edge behavior. | +| **`windowDrag.js`** | `makeWindowDraggable` | Drag support for floating panels. | +| **`modalSnap.js`**, **`toolWindowZOrder.js`**, **`windowResize.js`** | — | Modal snapping, z-index management, resize handles. | + +--- + +## 3. Chat Pipeline + +The largest and most central subsystem. Chat submission → backend SSE → progressive rendering of text, tools, research, documents, and UI events. + +| Module | Responsibility | +|---|---| +| **`chat.js`** | Main chat controller. Handles `handleChatSubmit`, stops/continues, builds `FormData`, posts to `/api/chat_stream`, reads the SSE stream, and dispatches each JSON event to the appropriate renderer. Tracks background streams, stalls, auto-recovery, and multi-round agent state. | +| **`chatStream.js`** | Helpers shared between streaming consumers: browser notifications, background-stream completion toasts, and `ui_control` event handling. | +| **`chatRenderer.js`** | Message DOM construction: `addMessage`, role labels, model route labels, color coding, footers, metrics, code blocks, sources boxes (`web`/`research`/`RAG`), findings box, images, report links, ask-user cards, welcome screen, and transcript utilities. | +| **`streamingRenderer.js`** | Incremental streaming renderer used by `chat.js`. Freezes finalized DOM blocks and only re-renders the growing tail to avoid flicker and O(N²) re-parsing. | +| **`streamingSegmenter.js`** | Splits a token stream into display units (text vs code fences) for `streamingRenderer.js`. | +| **`slashCommands.js`** | Slash-command registry (`/help`, `/setup`, etc.), parsing, and dispatch handlers. Exported functions are consumed by `chat.js` and `slashAutocomplete.js`. | +| **`slashAutocomplete.js`** | Composer autocomplete popup for `/` commands. | +| **`composerArrowUpRecall.js`** | Recall last user message with `↑` on an empty composer. | +| **`assistant.js`** | Assistant/persona behaviors and message styling helpers. | +| **`tts-ai.js`** | AI text-to-speech manager, enqueueing, streaming TTS, and playback button injection. | +| **`voiceRecorder.js`** | Voice recording from the composer microphone. | +| **`fileHandler.js`** | Attachment picker, paste/drop handling, upload, attachment strip rendering, pending-file management. | +| **`codeRunner.js`** | Client-side execution affordances for code blocks returned by the model. | + +--- + +## 4. Model, Endpoint, and Configuration Modules + +| Module | Responsibility | +|---|---| +| **`models.js`** | Model discovery / scanning, local model port probing, provider management, model selection UI state. | +| **`modelPicker.js`** | Composer model-picker dropdown and endpoint selection. | +| **`modelSort.js`** | Sorting helpers for model lists. | +| **`model/matchKey.js`** | Model-to-key matching helper. | +| **`providers.js`** | Provider metadata and account-management helpers. | +| **`providerDeviceFlow.js`** | OAuth device-flow support for providers. | +| **`presets.js`** | Character/preset selection, custom preset saving, inject prefix/suffix handling. | +| **`search.js`** | Web-search settings, provider selection, API key management. | +| **`settings.js`** | Settings panel (models, search, appearance, users, MCP, RAG, embedding, tokens). | +| **`admin.js`** | Admin panel and privileged user/endpoint configuration. | +| **`theme.js`** | Theme presets, custom colors, fonts, backgrounds, live theme switching. | + +--- + +## 5. Session, Sidebar, and Workspace + +| Module | Responsibility | +|---|---| +| **`sessions.js`** | Chat session list loading, creation, switching, renaming, archiving, library modal, and direct-chat creation. Tracks current session, streaming/research indicators in the sidebar. | +| **`workspace.js`** | Workspace folder path management for shell/file tool confinement. | +| **`search-chat.js`** | In-chat history search. | +| **`skills.js`** | Client-side skill library UI (load, edit, delete, test, and audit status display). | --- -## Core Modules (in static/js/) - -### 1. **ui.js** -- UI helper functions and utilities -- Toast notifications (`showToast`, `showError`) -- Element getter (`el()`) -- Clipboard operations (`copyToClipboard`) -- Scroll management (`scrollHistory`, `setAutoScroll`) -- Auto-resize textarea -- Debounce utility - -### 2. **markdown.js** -- Markdown processing and rendering -- Convert markdown to HTML (`mdToHtml`) -- Code block handling with syntax highlighting -- Content rendering for message arrays -- Text cleanup (`squashOutsideCode`) - -### 3. **sessions.js** -- Session/chat management -- Create, load, delete, switch sessions -- Session history loading -- Direct chat creation with models -- Session renaming - -### 4. **memory.js** -- AI memory management -- Load, add, edit, delete memories -- Memory search/filtering -- Memory UI rendering -- Memory count updates - -### 5. **fileHandler.js** -- File attachment handling -- File picker dialog -- File upload to server -- Attachment strip rendering -- Pending files management -- File preview/removal - -### 6. **voiceRecorder.js** -- Voice recording functionality -- Start/stop recording -- Audio file creation -- Microphone permission handling -- Recording UI updates - -### 7. **models.js** -- Model scanning and display -- Local model discovery (ports 8000-8020) -- Provider management (OpenAI) -- Model selection UI - -### 8. **rag.js** -- RAG (Retrieval Augmented Generation) management -- Load personal documents -- Add directories to RAG -- Display included files/directories - -### 9. **presets.js** -- Conversation preset management -- Load, save, activate presets -- Custom preset configuration -- Temperature, tokens, system prompt settings - -### 10. **search.js** -- Web search settings -- Provider selection (DuckDuckGo, Brave, SearXNG) -- API key management -- Save/load search configuration - -### 11. **chat.js** ⭐ (The Big One) -- Main chat functionality -- Message handling (`addMessage`) -- Chat submission (`handleChatSubmit`) -- Streaming response handling -- Performance metrics display -- Abort request management -- Loading states and error handling +## 6. Knowledge, Memory, and RAG + +| Module | Responsibility | +|---|---| +| **`memory.js`** | AI memory CRUD, search/filter UI, memory extraction, count badge. | +| **`rag.js`** | Personal document RAG: load documents, add directories/files, show included paths. | +| **`group.js`** | Group-chat UI and model orchestration. | + +--- + +## 7. Document and Editor Subsystems + +| Module | Responsibility | +|---|---| +| **`document.js`** | Tabbed document editor, AI edit suggestions, Markdown/HTML/CSV editing, document streaming (`streamDocOpen`/`streamDocDelta`), and panel state. | +| **`documentLibrary.js`** | Document library modal. | +| **`editor/`** | Gallery image editor canvas modules: layers, brush, inpaint, crop, filters, state, history panel, top-bar wiring, canvas coordinate helpers, and AI model runners for inpainting/background-removal. | + +--- + +## 8. Research UI + +| Module | Responsibility | +|---|---| +| **`research/panel.js`** | Research panel UI, job list, and controls. | +| **`research/jobs.js`** | Research job polling and status rendering. | +| **`researchSynapse.js`** | Animated research-progress visualization shown inside the chat bubble during a research run. | + +--- + +## 9. Gallery, Email, Calendar, Tasks, and Notes + +| Module | Responsibility | +|---|---| +| **`gallery.js`** / **`galleryEditor.js`** | Gallery/image library and canvas editor entry points. | +| **`emailInbox.js`** / **`emailLibrary.js`** | Email inbox reader and library modal. Sub-modules handle signatures, reply recipients, state, and signature folding. | +| **`calendar.js`** / **`calendar/utils.js`** / **`calendar/reminders.js`** | Calendar views, event forms, reminders. | +| **`tasks.js`** | Scheduled task/recurring LLM job UI. | +| **`notes.js`** | Notes and todo panel, reminders, pinboard. | + +--- + +## 10. Cookbook (Model Serving) + +| Module | Responsibility | +|---|---| +| **`cookbook.js`** | Cookbook main UI: hardware fitting, presets, action panels. | +| **`cookbook-hwfit.js`** / **`cookbook-diagnosis.js`** / **`cookbook-deps-recipes.js`** | Hardware-fit scoring, dependency diagnosis, recipe handling. | +| **`cookbookDownload.js`** / **`cookbookServe.js`** / **`cookbookRunning.js`** / **`cookbookSchedule.js`** / **`cookbookPorts.js`** / **`cookbookProgressSignal.js`** | Model download/serve flow, running job cards, scheduling, port detection, and progress computation. | --- -## Main Application File +## 11. Compare and Utility Modules + +| Module | Responsibility | +|---|---| +| **`compare/index.js`** (with `compare/state.js`, `compare/stream.js`, `compare/panes.js`, `compare/selector.js`, `compare/scoreboard.js`, `compare/probe.js`, `compare/vote.js`, `compare/icons.js`) | Model compare mode: parallel streams, panes, scoring, vote UI. | +| **`censor.js`** | Text/image censor overlay toggles. | +| **`a11y.js`** | Accessibility helpers. | +| **`platform.js`** | Platform detection (macOS/Windows/Linux) and keyboard-modifier helpers. | +| **`escMenuStack.js`** | Stack manager for dismissible popups. | +| **`dragSort.js`** | Drag-to-sort shared behavior. | +| **`tourHints.js`** / **`tourAutoplay.js`** | Onboarding tour helpers. | +| **`color/hex.js`**, **`colorPicker.js`**, **`langIcons.js`**, **`util/ordinal.js`** | Small utility modules for color, language icons, and formatting. | + +--- -### **app.js** -- Application initialization -- Event listener setup -- Drag & drop handlers -- Keyboard shortcuts -- Module initialization -- Global configuration (API_BASE) -- Coordinates all modules together +## 12. Frontend Event Streaming Flow + +``` +User submits composer + └── chat.js::handleChatSubmit() builds FormData + ├── fileHandler.uploadPending() for attachments + ├── document.js saved (if a document panel is open) + └── POST /api/chat_stream + +Server responds with SSE stream + └── chat.js reads chunks via res.body.getReader() + TextDecoder + ├── Lines starting with "event:" set next-error state + └── Lines starting with "data:" carry JSON payloads + +JSON events are dispatched by "type": + delta → streamingRenderer → markdown → live reply text + agent_prep → update spinner label + tool_start → finalize text bubble; create agent-thread node with wave animation + tool_progress → append/update live stdout/stderr tail + tool_output → mark node done/failed, render output, diffs, screenshots + agent_step → finalize tool thread; create new msg-continuation bubble + doc_stream_open → document.js opens a live document + doc_stream_delta → document.js appends content to that document + research_progress → researchSynapse visualization + spinner timer + research_sources → build sources box for research + research_done → reload session history to show the report + web_sources → build web-search sources box + model_info → update role header with requested/actual model + fallback → show fallback model toast + update role label + metrics → collect/display token/cost metrics + message_saved → store database id on the message element + budget_exceeded → show budget banner + rounds_exhausted → show Continue button for step-limit hits + teacher_takeover → insert escalation banner, reset round state + skill_saved → show skill-learned banner +``` + +Foreground vs background streams: +- If the user switches sessions while a stream is running, `chat.js` pauses DOM + updates and stores the state in `_backgroundStreams`. Completion is signaled + with a sidebar dot/notifications, and the history is reloaded when the user + returns. --- -## Dependency Order (Load Order in HTML) -```html - - - - - - - - - - - - +## 13. What Changed from the Previous Summary + +- The frontend is now exclusively ES6-module based; the old `