diff --git a/navi/api/routes/sessions.py b/navi/api/routes/sessions.py deleted file mode 100644 index 6194ee1..0000000 --- a/navi/api/routes/sessions.py +++ /dev/null @@ -1,1064 +0,0 @@ -"""Session management endpoints.""" - -import asyncio -from datetime import datetime, timedelta, timezone -from typing import Annotated, Any - -import mimetypes - -import structlog -from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile -from fastapi.responses import FileResponse -from pydantic import BaseModel - -from navi.api.deps import ( - get_backend_registry, - get_kv_store, - get_memory_store, - get_orchestrator, - get_profile_registry, - get_scheduler, - get_session_store, - get_todo_store, - require_admin, - require_user, -) -from navi.auth import User -from navi.auth.deps import check_session_access -from navi.content_store import list_for_session -from navi.content_types import detect, is_inline_mime -from navi.config import settings -from navi.core import BackendRegistry, ProfileRegistry, SessionStore -from navi.core.scheduler import RecallScheduler -from navi.core.name_generator import generate_session_name -from navi.exceptions import ProfileNotFound -from navi.memory import MemoryStore -from navi.profiles.base import admin_only_blocked -from navi.session_files import delete_session_dir, ensure_session_dir, is_forbidden, list_session_files, safe_filename, session_dir -from navi.store import KvStore -from navi.todo_store import PgTodoStore - -log = structlog.get_logger() - -router = APIRouter(prefix="/sessions", tags=["sessions"]) - -class CreateSessionRequest(BaseModel): - profile_id: str | None = None - - -class PinSessionRequest(BaseModel): - pinned: bool - - -class RenameSessionRequest(BaseModel): - # Empty string means "regenerate from session content" (the client sends - # it when the user cleared the title and left the edit field). - name: str = "" - - -def _session_owner(user) -> str | None: - """Owner id for session create/list. - - In local mode (auth disabled) the anonymous user is a separate user with - user_id=None — not an alias of every other user — so sessions must be - persisted ownerless and listings must scope to NULL-owner rows only. - """ - return user.id if settings.navi_auth_enabled else None - - -def _session_admin(user) -> bool: - """Listing-scope admin flag — only meaningful when auth is enabled. - - The local-mode anonymous user is nominally 'admin' so debug/admin - endpoints keep working, but the admin flag must never widen the chat - session listings beyond NULL-owner rows there. - """ - return settings.navi_auth_enabled and ( - user.role == "admin" or user.has_permission("navi.sessions.read_all") - ) - - -def _profile_of(profiles: ProfileRegistry, profile_id: str | None): - """Profile or None — a deleted profile must not break the session head.""" - if not profile_id: - return None - try: - return profiles.get(profile_id) - except ProfileNotFound: - return None - - -def _backend_for(profile, backends: BackendRegistry): - key = getattr(profile, "llm_backend", "ollama") if profile is not None else "ollama" - try: - return backends.get(key) - except KeyError: - return None - - -def _effective_window(profile, backend) -> int: - """Context window the agent holds this profile's prompts to. - - Same precedence as ``Agent._window_for``: the route's active entry wins, - then the profile's own value, then the global default. The session head - used to report the global ``OLLAMA_NUM_CTX`` even when the route serving - the session declared another window — the client then drew a percentage - against a denominator the agent never used. - """ - if profile is not None: - active_window = getattr(backend, "active_window", None) - if callable(active_window): - try: - window = active_window(getattr(profile, "llm_route", None)) - except Exception: # pragma: no cover - a backend may not know profiles - window = None - if window: - return window - own = getattr(profile, "context_window", None) - if own: - return own - return settings.ollama_num_ctx - - -def _llm_info(profile, backend) -> dict: - """What the route engine is doing for this session's profile. - - In-process state only (the background probe keeps it current): the active - entry — model, target, host, window — and the chain with each target's - health. ``configured: false`` means the deployment has no ``llm.json`` - routes; the profile's model list is then all there is to show. - """ - route_id = getattr(profile, "llm_route", None) if profile is not None else None - models = list(getattr(profile, "model", None) or []) if profile is not None else [] - info: dict[str, Any] = { - "configured": False, - "route": route_id, - "models": models, - "active": None, - "chain": [], - } - if backend is None or not hasattr(backend, "active_report"): - return info - - info["configured"] = True - config = getattr(backend, "config", None) - route = config.route(route_id) if config is not None else None - report = backend.active_report() or {} - active = report.get(route_id) if route_id else None - - states: dict[str, dict] = {} - missing: set[tuple[str, str]] = set() - health = getattr(backend, "health", None) - if health is not None: - snapshot = health.snapshot() - states = snapshot.get("targets", {}) - missing = {(m["target"], m["model"]) for m in snapshot.get("missing_models", [])} - - chain: list[dict] = [] - if route is not None: - for entry in route.entries: - state = states.get(entry.target.name, {}) - item = { - "model": entry.model, - "target": entry.target.name, - "host": entry.target.host, - "primary": entry is route.primary, - "state": "missing" if (entry.target.name, entry.model) in missing - else state.get("state", "ok"), - } - if state.get("reason"): - item["reason"] = state["reason"] - chain.append(item) - info["chain"] = chain - - if active is not None: - host = next( - (e["host"] for e in chain - if e["model"] == active["model"] and e["target"] == active["target"]), - "", - ) - info["active"] = { - "model": active["model"], - "target": active["target"], - "host": host, - "primary": bool(active.get("primary", True)), - "context_window": _effective_window(profile, backend), - } - return info - - -@router.post("", status_code=201) -async def create_session( - payload: CreateSessionRequest, - store: Annotated[SessionStore, Depends(get_session_store)], - profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], - memory: Annotated[MemoryStore, Depends(get_memory_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - profile_id = payload.profile_id - if not profile_id: - profile_id = settings.navi_default_profile_id - if not profile_id: - raise HTTPException(status_code=400, detail="profile_id is required and no default profile is configured") - - try: - profile = profiles.get(profile_id) - except ProfileNotFound: - raise HTTPException(status_code=404, detail=f"Profile '{profile_id}' not found") - - if admin_only_blocked(profile, user.role): - raise HTTPException( - status_code=403, detail=f"Profile '{profile_id}' requires admin access" - ) - - session = await store.create(profile_id, user_id=_session_owner(user)) - - # Fire-and-forget: extract memory from any stale sessions (last active > 30 min ago) - asyncio.create_task(_process_stale_sessions(store, memory, user_id=_session_owner(user))) - - return { - "session_id": session.id, - "profile_id": session.profile_id, - "created_at": session.created_at.isoformat(), - } - - -async def _process_stale_sessions(store: SessionStore, memory: MemoryStore, user_id: str | None = None) -> None: - """Extract facts from sessions idle for >30 min that haven't been processed yet.""" - from navi.api.deps import get_backend_registry - from navi.config import settings - from navi.memory.extractor import extract_and_update - - cutoff = datetime.now(timezone.utc) - timedelta(minutes=30) - try: - sessions = await store.list_all(user_id=user_id, is_admin=False) - backend = get_backend_registry().get("ollama") - except Exception: - return - - for session in sessions: - if not session.messages: - continue - if session.user_id is None: - continue # skip legacy sessions — no multi-user memory for unowned sessions - if session.last_active >= cutoff: - continue # still active - extracted_at = await memory.get_extracted_at(session.id) - if extracted_at and extracted_at >= session.last_active.isoformat(): - continue # already up to date - try: - await extract_and_update(session, backend, settings.ollama_default_model, memory) - except Exception: - log.warning("memory.extraction_failed", session_id=session.id, exc_info=True) - - -@router.get("") -async def list_sessions( - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], - scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], - limit: Annotated[int | None, Query(ge=1, le=100)] = None, - offset: Annotated[int | None, Query(ge=0)] = None, - profile_id: str | None = None, - search: str | None = Query(None), - special: bool | None = Query(None), -) -> dict | list[dict]: - # special=None and special=False both mean the regular list; special=True - # exposes the service sessions (e.g. Synapse reactions), which the sidebar - # hides behind its toggle. - special_filter = False if special is None else special - is_admin = _session_admin(user) - - if limit is None and offset is None: - sessions = await store.list_all(user_id=_session_owner(user), is_admin=is_admin, special=special_filter) - if profile_id and not search: - sessions = [s for s in sessions if s.profile_id == profile_id] - pending_ids = await scheduler.get_pending_session_ids([s.id for s in sessions]) - return [_session_summary(s, search=search, has_pending_recall=s.id in pending_ids) for s in sessions] - - page_limit = limit or 30 - page_offset = offset or 0 - if search: - sessions = await store.search_list( - limit=page_limit + 1, - offset=page_offset, - user_id=_session_owner(user), - is_admin=is_admin, - search=search, - special=special_filter, - ) - else: - sessions = await store.list_page( - limit=page_limit + 1, - offset=page_offset, - profile_id=profile_id, - user_id=_session_owner(user), - is_admin=is_admin, - special=special_filter, - ) - items = sessions[:page_limit] - pending_ids = await scheduler.get_pending_session_ids([s.id for s in items]) - return { - "items": [_session_summary(s, search=search, has_pending_recall=s.id in pending_ids) for s in items], - "limit": page_limit, - "offset": page_offset, - "has_more": len(sessions) > page_limit, - "next_offset": page_offset + len(items), - } - - -def _session_summary(session, search: str | None = None, has_pending_recall: bool = False) -> dict: - summary = { - "session_id": session.id, - "profile_id": session.profile_id, - "name": session.name, - "message_count": len(session.messages), - "preview": _preview(session), - "pinned": session.pinned, - "special": session.special, - "has_pending_recall": has_pending_recall, - "created_at": session.created_at.isoformat(), - "last_active": session.last_active.isoformat(), - } - if search: - summary["match_indices"], summary["match_preview"] = _match_info(session, search) - return summary - - -def _preview(session) -> str: - """Return a short snippet from the last user or assistant message.""" - for msg in reversed(session.messages): - if msg.role in ("user", "assistant") and msg.content: - return msg.content[:60] - return "" - - -def _match_info(session, search: str) -> tuple[list[int], str]: - """Return message indices that contain the search term and a preview snippet.""" - q = search.lower() - indices = [] - preview = "" - for i, msg in enumerate(session.messages): - content = msg.content or "" - if q in content.lower(): - indices.append(i) - if not preview: - idx = content.lower().find(q) - start = max(0, idx - 30) - end = min(len(content), idx + len(search) + 30) - snippet = content[start:end] - if start > 0: - snippet = "…" + snippet - if end < len(content): - snippet = snippet + "…" - preview = snippet - return indices, preview - - -@router.get("/{session_id}/meta") -async def get_session_meta( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], - backends: Annotated[BackendRegistry, Depends(get_backend_registry)], - user: Annotated[User, Depends(require_user)], -) -> dict: - # Light session head without the message payload — the webclient loads the - # history separately via /{session_id}/messages/history in pages. - session = await store.get_meta(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - profile = _profile_of(profiles, session.profile_id) - backend = _backend_for(profile, backends) - return { - "session_id": session.id, - "profile_id": session.profile_id, - "name": session.name, - "message_count": session.db_message_count, - "context_token_count": session.context_token_count, - "max_context_tokens": _effective_window(profile, backend), - "archive_threshold": session.archive_threshold, - "created_at": session.created_at.isoformat(), - "last_active": session.last_active.isoformat(), - } - - -def _recall_payload(recall: Any) -> dict | None: - """The recall as the client reads it — one shape everywhere.""" - if recall is None: - return None - return { - "id": recall.id, - "call_type": recall.call_type, - "trigger_at": recall.trigger_at.isoformat(), - "interval_seconds": recall.interval_seconds, - "internal_comment": recall.internal_comment, - "additional_context_message": recall.additional_context_message, - "status": recall.status, - "created_at": recall.created_at.isoformat(), - } - - -@router.get("/{session_id}/info") -async def get_session_info( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], - backends: Annotated[BackendRegistry, Depends(get_backend_registry)], - orchestrator: Annotated[Any, Depends(get_orchestrator)], - scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """Everything the session-info dialog shows, in one authenticated call. - - The model identity lives in the route engine, not in the session row, so - this joins the session head with the profile's route: the active entry - (model, target, host, window) and the health of every entry in the chain. - Reads in-process state only — the background probe keeps it current. - """ - session = await store.get_meta(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - - profile = _profile_of(profiles, session.profile_id) - backend = _backend_for(profile, backends) - - keep_recent = getattr(profile, "compression_keep_recent", None) if profile else None - summary_max = getattr(profile, "compression_max_tokens", None) if profile else None - - return { - "session_id": session.id, - "name": session.name, - "profile_id": session.profile_id, - "profile_name": getattr(profile, "name", None) if profile else None, - "created_at": session.created_at.isoformat(), - "last_active": session.last_active.isoformat(), - "message_count": session.db_message_count, - "context_token_count": session.context_token_count, - "max_context_tokens": _effective_window(profile, backend), - "archive_threshold": session.archive_threshold, - "compression": { - "enabled": settings.context_compression_enabled, - "threshold": settings.context_compression_threshold, - "keep_recent": keep_recent or settings.context_keep_recent, - "summary_max_tokens": summary_max or settings.context_summary_max_tokens, - }, - "llm": _llm_info(profile, backend), - "live": { - "running": orchestrator.is_running(session_id), - "clients": orchestrator.client_count(session_id), - }, - "recall": _recall_payload(await scheduler.get_pending_recall(session_id)), - } - - -@router.get("/{session_id}") -async def get_session( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], - backends: Annotated[BackendRegistry, Depends(get_backend_registry)], - user: Annotated[User, Depends(require_user)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - profile = _profile_of(profiles, session.profile_id) - backend = _backend_for(profile, backends) - return { - "session_id": session.id, - "profile_id": session.profile_id, - "name": session.name, - "messages": [m.model_dump(mode='json', exclude_none=True) for m in session.messages], - "context_token_count": session.context_token_count, - "max_context_tokens": _effective_window(profile, backend), - "archive_threshold": session.archive_threshold, - "created_at": session.created_at.isoformat(), - "last_active": session.last_active.isoformat(), - } - - -@router.get("/{session_id}/todos") -async def get_session_todos( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - kv: Annotated[KvStore, Depends(get_kv_store)], - todos: Annotated[PgTodoStore, Depends(get_todo_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """Return the session's current todo list (the parent agent's plan). - - Reads the parent session's todo row via explicit (user_id, session_id) — no - ContextVars. Sub-agent todos live in isolated rows and are not exposed here - (Phase 3). Used by the TUI to seed the side panel on attach/switch. - """ - from navi.tools.todo import load_tasks_for - - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - tasks = await load_tasks_for(session_id, session.user_id, kv=kv, store=todos) - return {"session_id": session_id, "tasks": tasks} - - -@router.get("/{session_id}/terminals") -async def list_session_terminals( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """List the session's active persistent terminals (background dev servers, - watchers, …). Seeds the TUI terminals modal + status-bar count on attach.""" - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - - from navi.api.deps import get_orchestrator - - orchestrator = get_orchestrator() - terminals: list[dict] = [] - if orchestrator and orchestrator._container and orchestrator._container.terminal_manager: - terminals = orchestrator._container.terminal_manager.list(session_id) - return {"session_id": session_id, "terminals": terminals} - - -@router.post("/{session_id}/terminals/{terminal_name}/close") -async def close_session_terminal( - session_id: str, - terminal_name: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """Force-close one persistent terminal by name (from the TUI /terminals modal).""" - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - - from navi.api.deps import get_orchestrator - - orchestrator = get_orchestrator() - if not (orchestrator and orchestrator._container and orchestrator._container.terminal_manager): - return {"session_id": session_id, "terminal_name": terminal_name, "closed": False} - closed = await orchestrator._container.terminal_manager.close(session_id, terminal_name) - return {"session_id": session_id, "terminal_name": terminal_name, "closed": closed} - - -@router.patch("/{session_id}/pin") -async def pin_session( - session_id: str, - body: PinSessionRequest, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - ok = await store.set_pinned(session_id, body.pinned) - if not ok: - raise HTTPException(status_code=404, detail="Session not found") - return {"session_id": session_id, "pinned": body.pinned} - - -@router.get("/{session_id}/context") -async def get_session_context( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_admin)], -) -> dict: - """Return the LLM context (what the model actually sees) for debugging.""" - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - total_chars = sum(len(m.content or "") for m in session.context) - return { - "session_id": session.id, - "profile_id": session.profile_id, - "message_count": len(session.context), - "total_chars": total_chars, - "context": [m.model_dump(mode="json", exclude_none=True) for m in session.context], - } - - -@router.get("/{session_id}/planning") -async def get_session_planning( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_admin)], -) -> dict: - """Return all planning phase debug logs for a session.""" - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - return {"session_id": session.id, "logs": session.planning_logs} - - -@router.get("/{session_id}/content") -async def get_session_content( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], -) -> dict: - """Return published inline content records for this session. - - Public — no auth required. The session ID acts as an unguessable - capability token for accessing shared/published files. - """ - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - return {"session_id": session.id, "content": await list_for_session(session_id)} - - -@router.post("/{session_id}/files", status_code=201) -async def upload_file( - session_id: str, - file: UploadFile, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - - name = safe_filename(file.filename or "upload") - if is_forbidden(name): - raise HTTPException(status_code=400, detail="Executable files are not allowed") - - max_bytes = settings.session_files_max_size_mb * 1024 * 1024 - dest = ensure_session_dir(session_id) / name - - # Handle duplicate filenames - if dest.exists(): - stem, suffix = dest.stem, dest.suffix - i = 1 - while dest.exists(): - dest = dest.with_name(f"{stem}_{i}{suffix}") - i += 1 - - # Stream chunks directly to disk to avoid buffering the whole file in RAM - size = 0 - try: - with dest.open("wb") as f: - while True: - chunk = await file.read(1024 * 1024) # 1 MB at a time - if not chunk: - break - size += len(chunk) - if size > max_bytes: - raise HTTPException( - status_code=413, - detail=f"File exceeds {settings.session_files_max_size_mb} MB limit", - ) - await asyncio.to_thread(f.write, chunk) - except HTTPException: - raise - except Exception as exc: - await asyncio.to_thread(dest.unlink, True) - raise HTTPException(status_code=500, detail=f"Upload failed: {exc}") from exc - - return { - "name": dest.name, - "size": size, - "path": str(dest), - "content_type": file.content_type or "application/octet-stream", - } - - -@router.get("/{session_id}/files") -async def list_session_files_endpoint( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """Return all files and directories in the session's file directory (recursive, depth 10).""" - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.files.read_all") - - files = list_session_files(session_id) - return {"session_id": session_id, "files": files} - - -@router.get("/{session_id}/files/{filename}") -async def download_file( - session_id: str, - filename: str, - store: Annotated[SessionStore, Depends(get_session_store)], - download: bool = False, -) -> FileResponse: - """Download a file from the session's file directory. - - Public — no auth required. The session ID acts as an unguessable - capability token for accessing shared/published files. - """ - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - - # Resolve and verify the file is within the session directory (no path traversal) - base = session_dir(session_id).resolve() - file_path = (base / filename).resolve() - try: - file_path.relative_to(base) - except ValueError: - raise HTTPException(status_code=403, detail="Access denied") - - if not file_path.exists() or not file_path.is_file(): - raise HTTPException(status_code=404, detail="File not found") - - # The session's own registry decides the type — the host's /etc/mime.types - # is the fallback for formats it does not cover (docx, archives), where a - # sensible MIME still makes the saved file openable. - _, content_type = detect(file_path.name, file_path) - if content_type == "application/octet-stream": - content_type = mimetypes.guess_type(file_path.name)[0] or content_type - # Video and audio belong in this list as much as images do: served as an - # attachment they are downloaded instead of played, even when the client - # asks the browser to open the URL rather than the in-page player. - inline = not download and is_inline_mime(content_type) - - return FileResponse( - path=file_path, - filename=filename, - media_type=content_type, - content_disposition_type="inline" if inline else "attachment", - ) - - -@router.get("/{session_id}/tasks") -async def list_session_tasks( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """List this session's background tasks (the TaskManager jobs). - - task_update events are delivered live over the WS and are not replayed on - reconnect, so the client fetches this snapshot to rebuild its - backgrounds tab after a reload / reconnection. - """ - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.read_all") - - from datetime import datetime, timezone as _tz - - from navi.core.tasks import args_summary, get_task_manager - - items = [] - for j in get_task_manager().list(session_id): - items.append({ - "task_id": j.task_id, - "tool": j.tool, - "status": j.status, - "args_summary": args_summary(j.args), - "parent_tool_call_id": j.parent_tool_call_id or None, - "started_at": datetime.fromtimestamp(j.created_at, tz=_tz.utc).isoformat(), - "finished_at": ( - datetime.fromtimestamp(j.finished_at, tz=_tz.utc).isoformat() - if j.finished_at is not None else None - ), - "subagent_tokens": j.subagent_tokens, - "result_preview": j.preview() or None, - }) - return {"session_id": session_id, "tasks": items} - - -@router.post("/{session_id}/generate-name") -async def generate_name( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - backends: Annotated[BackendRegistry, Depends(get_backend_registry)], - user: Annotated[User, Depends(require_user)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - - # Only generate if no name yet - if session.name: - return {"name": session.name} - - user_messages = [ - m.content for m in session.messages - if m.role == "user" and m.content - ] - if not user_messages: - return {"name": None} - - # Short first user messages often carry no topic (greetings, one-liners) — - # that's why a share of sessions stayed unnamed. The agent's final replies - # are where the actual subject lands; fold them into the naming context. - assistant_messages = [ - m.content for m in session.messages - if m.role == "assistant" and m.content - ][-2:] - - backend = backends.get("ollama") - name = await generate_session_name( - user_messages, backend, settings.ollama_default_model, - assistant_messages=assistant_messages, - ) - - if name: - await store.set_name(session_id, name) - - return {"name": name} - - -@router.patch("/{session_id}/name") -async def rename_session( - session_id: str, - body: RenameSessionRequest, - store: Annotated[SessionStore, Depends(get_session_store)], - backends: Annotated[BackendRegistry, Depends(get_backend_registry)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """Set the session title manually, or regenerate it on an empty edit. - - A whitespace-only name falls back to LLM naming (the same context as - generate-name: user messages plus the agent's final replies). - """ - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - - name = body.name.strip()[:200] - if not name: - user_messages = [ - m.content for m in session.messages - if m.role == "user" and m.content - ] - assistant_messages = [ - m.content for m in session.messages - if m.role == "assistant" and m.content - ][-2:] - if not user_messages: - return {"name": None} - backend = backends.get("ollama") - name = await generate_session_name( - user_messages, backend, settings.ollama_default_model, - assistant_messages=assistant_messages, - ) - if not name: - return {"name": None} - - ok = await store.set_name(session_id, name) - if not ok: - raise HTTPException(status_code=404, detail="Session not found") - return {"session_id": session_id, "name": name} - - -@router.delete("/{session_id}", status_code=204) -async def delete_session( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> None: - import time as _time - - _t0 = _time.perf_counter() - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user, permission="navi.sessions.delete_all") - # Close any persistent terminals tied to this session before deleting - from navi.api.deps import get_orchestrator - orchestrator = get_orchestrator() - if orchestrator and orchestrator._container and orchestrator._container.terminal_manager: - await orchestrator._container.terminal_manager.close_all(session_id) - - # Cancel background tasks for this session: they outlive the run but not - # their session, and letting them finish would burn tokens for a session - # that no longer exists while their completion notes dangle in the KV - # store (suppress_note = True, set inside cancel_session_jobs). - from navi.core.tasks import get_task_manager - from navi.core.task_notes import clear as clear_task_notes - _t_get = _time.perf_counter() - get_task_manager().cancel_session_jobs(session_id) - # Pending notes from earlier completed jobs would never drain — the - # session is going away. - await clear_task_notes(session_id) - _t_tasks = _time.perf_counter() - - deleted = await store.delete(session_id) - _t_db = _time.perf_counter() - if not deleted: - raise HTTPException(status_code=404, detail="Session not found") - # Remove session file directory if it exists - await delete_session_dir(session_id) - _t_dir = _time.perf_counter() - log.info( - "sessions.delete_timing", - session_id=session_id, - lookup_terminals_ms=round((_t_get - _t0) * 1000, 1), - tasks_notes_ms=round((_t_tasks - _t_get) * 1000, 1), - db_delete_ms=round((_t_db - _t_tasks) * 1000, 1), - files_dir_ms=round((_t_dir - _t_db) * 1000, 1), - total_ms=round((_t_dir - _t0) * 1000, 1), - ) - - -def _require_share_owner(session, user: User) -> None: - """Only the owner (or an admin) may hand out a chat's history. - - Deliberately stricter than ``check_session_access``: a read-all permission - lets staff *read* other users' sessions, not distribute them. - """ - if not settings.navi_auth_enabled: - if session.user_id is not None: - raise HTTPException(status_code=403, detail="Access denied") - return - if session.user_id is not None and session.user_id == user.id: - return - if user.role == "admin": - return - raise HTTPException(status_code=403, detail="Access denied") - - -@router.post("/{session_id}/share") -async def share_session( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], -) -> dict: - """Issue (or re-issue) the session's public read-only link. - - The link is a capability: whoever holds it reads the display history and - nothing else — no input, no session list, no artifacts panel. It has no - expiry and no revoke switch by decision; deleting the session closes it. - One share per session, so a second call hands back the same link instead - of multiplying exposure. - """ - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - _require_share_owner(session, user) - - share = await store.get_or_create_share(session_id, session.user_id) - if share is None: - # Lazily-persisted session: no row yet, so there is nothing to point at. - raise HTTPException(status_code=409, detail="Session has no saved messages yet") - - token = share["token"] - log.info("sessions.shared", session_id=session_id, user_id=user.id) - return { - "token": token, - "url": f"{settings.public_url.rstrip('/')}/#share={token}", - "created_at": share["created_at"].isoformat(), - } - - -@router.get("/{session_id}/messages/archive") -async def get_session_archived_messages( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], - before_seq: int | None = None, - limit: int = Query(50, ge=1, le=200), -) -> dict: - """Return older archived messages for scroll-up loading.""" - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - msgs = await store.get_archived_messages(session_id, before_seq=before_seq, limit=limit) - next_before_seq = msgs[0].sequence_number if msgs else None - return { - "items": [m.model_dump(mode="json", exclude_none=True) for m in msgs], - "has_more": len(msgs) == limit, - "next_before_seq": next_before_seq, - } - - -@router.get("/{session_id}/messages/history") -async def get_session_history( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], - before_seq: int | None = None, - limit: int = Query(150, ge=1, le=200), -) -> dict: - """Paged full history (archive ∪ hot), newest page first. - - Without ``before_seq`` — the tail a client shows initially; with it — the - next older page for scroll-up loading. Returns oldest-first items so the - client can render/prepend them directly. - """ - session = await store.get_meta(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - msgs, has_more = await store.get_history_page(session_id, before_seq=before_seq, limit=limit) - return { - "items": [m.model_dump(mode="json", exclude_none=True) for m in msgs], - "has_more": has_more, - "next_before_seq": msgs[0].sequence_number if msgs else None, - } - - -@router.get("/{session_id}/recall") -async def get_session_recall( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], - scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - return {"recall": _recall_payload(await scheduler.get_pending_recall(session_id))} - - -@router.delete("/{session_id}/recall") -async def cancel_session_recall( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], - scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - ok = await scheduler.cancel_recall(session_id) - if ok: - from navi.core.event_bus import get_event_bus - from navi.core.events import RecallUpdate - await get_event_bus().publish( - RecallUpdate(session_id=session_id, status="cancelled", action="cancelled") - ) - return {"ok": ok} - - -@router.post("/{session_id}/recall/skip") -async def skip_session_recall( - session_id: str, - store: Annotated[SessionStore, Depends(get_session_store)], - user: Annotated[User, Depends(require_user)], - scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], -) -> dict: - session = await store.get(session_id) - if session is None: - raise HTTPException(status_code=404, detail="Session not found") - check_session_access(session, user) - ok = await scheduler.skip_next_recall(session_id) - if ok: - from navi.core.event_bus import get_event_bus - from navi.core.events import RecallUpdate - await get_event_bus().publish( - RecallUpdate(session_id=session_id, status="pending", action="rescheduled") - ) - return {"ok": ok} diff --git a/navi/api/routes/sessions/__init__.py b/navi/api/routes/sessions/__init__.py new file mode 100644 index 0000000..0031a83 --- /dev/null +++ b/navi/api/routes/sessions/__init__.py @@ -0,0 +1,73 @@ +"""Session management endpoints. + +One `/sessions` router composed from area modules: + +* `common` — request models and helpers shared by the endpoints; +* `listing` — creating and listing sessions; +* `detail` — the session head and read-only debug views; +* `files` — upload, listing and download of session files; +* `messages` — paged display history (archive and hot); +* `naming` — generated and manual titles; +* `manage` — pin, delete, share, todos, terminals, background tasks; +* `recall` — scheduled recalls attached to a session. + +The view functions are re-exported here so callers that imported them from the +former single module keep working. +""" + +from ._router import router +from .common import CreateSessionRequest, PinSessionRequest, RenameSessionRequest +from .detail import ( + get_session, + get_session_content, + get_session_context, + get_session_info, + get_session_meta, + get_session_planning, +) +from .files import download_file, list_session_files_endpoint, upload_file +from .listing import create_session, list_sessions +from .manage import ( + close_session_terminal, + delete_session, + get_session_todos, + list_session_tasks, + list_session_terminals, + pin_session, + share_session, +) +from .messages import get_session_archived_messages, get_session_history +from .naming import generate_name, rename_session +from .recall import cancel_session_recall, get_session_recall, skip_session_recall + +__all__ = [ + "CreateSessionRequest", + "PinSessionRequest", + "RenameSessionRequest", + "cancel_session_recall", + "close_session_terminal", + "create_session", + "delete_session", + "download_file", + "generate_name", + "get_session", + "get_session_archived_messages", + "get_session_content", + "get_session_context", + "get_session_history", + "get_session_info", + "get_session_meta", + "get_session_planning", + "get_session_recall", + "get_session_todos", + "list_session_files_endpoint", + "list_session_tasks", + "list_session_terminals", + "list_sessions", + "pin_session", + "rename_session", + "router", + "share_session", + "skip_session_recall", + "upload_file", +] diff --git a/navi/api/routes/sessions/_router.py b/navi/api/routes/sessions/_router.py new file mode 100644 index 0000000..0e36175 --- /dev/null +++ b/navi/api/routes/sessions/_router.py @@ -0,0 +1,5 @@ +"""The shared `/sessions` router — every endpoint module registers on it.""" + +from fastapi import APIRouter + +router = APIRouter(prefix="/sessions", tags=["sessions"]) diff --git a/navi/api/routes/sessions/common.py b/navi/api/routes/sessions/common.py new file mode 100644 index 0000000..eeadb89 --- /dev/null +++ b/navi/api/routes/sessions/common.py @@ -0,0 +1,186 @@ +"""Request models and shared helpers — owner scoping, profile/route info, payloads.""" + +from typing import Any + +from fastapi import HTTPException +from pydantic import BaseModel + +from navi.auth import User +from navi.config import settings +from navi.core import BackendRegistry, ProfileRegistry +from navi.exceptions import ProfileNotFound + + +class CreateSessionRequest(BaseModel): + profile_id: str | None = None + + +class PinSessionRequest(BaseModel): + pinned: bool + + +class RenameSessionRequest(BaseModel): + # Empty string means "regenerate from session content" (the client sends + # it when the user cleared the title and left the edit field). + name: str = "" + +def _session_owner(user) -> str | None: + """Owner id for session create/list. + + In local mode (auth disabled) the anonymous user is a separate user with + user_id=None — not an alias of every other user — so sessions must be + persisted ownerless and listings must scope to NULL-owner rows only. + """ + return user.id if settings.navi_auth_enabled else None + + +def _session_admin(user) -> bool: + """Listing-scope admin flag — only meaningful when auth is enabled. + + The local-mode anonymous user is nominally 'admin' so debug/admin + endpoints keep working, but the admin flag must never widen the chat + session listings beyond NULL-owner rows there. + """ + return settings.navi_auth_enabled and ( + user.role == "admin" or user.has_permission("navi.sessions.read_all") + ) + + +def _profile_of(profiles: ProfileRegistry, profile_id: str | None): + """Profile or None — a deleted profile must not break the session head.""" + if not profile_id: + return None + try: + return profiles.get(profile_id) + except ProfileNotFound: + return None + + +def _backend_for(profile, backends: BackendRegistry): + key = getattr(profile, "llm_backend", "ollama") if profile is not None else "ollama" + try: + return backends.get(key) + except KeyError: + return None + + +def _effective_window(profile, backend) -> int: + """Context window the agent holds this profile's prompts to. + + Same precedence as ``Agent._window_for``: the route's active entry wins, + then the profile's own value, then the global default. The session head + used to report the global ``OLLAMA_NUM_CTX`` even when the route serving + the session declared another window — the client then drew a percentage + against a denominator the agent never used. + """ + if profile is not None: + active_window = getattr(backend, "active_window", None) + if callable(active_window): + try: + window = active_window(getattr(profile, "llm_route", None)) + except Exception: # pragma: no cover - a backend may not know profiles + window = None + if window: + return window + own = getattr(profile, "context_window", None) + if own: + return own + return settings.ollama_num_ctx + + +def _llm_info(profile, backend) -> dict: + """What the route engine is doing for this session's profile. + + In-process state only (the background probe keeps it current): the active + entry — model, target, host, window — and the chain with each target's + health. ``configured: false`` means the deployment has no ``llm.json`` + routes; the profile's model list is then all there is to show. + """ + route_id = getattr(profile, "llm_route", None) if profile is not None else None + models = list(getattr(profile, "model", None) or []) if profile is not None else [] + info: dict[str, Any] = { + "configured": False, + "route": route_id, + "models": models, + "active": None, + "chain": [], + } + if backend is None or not hasattr(backend, "active_report"): + return info + + info["configured"] = True + config = getattr(backend, "config", None) + route = config.route(route_id) if config is not None else None + report = backend.active_report() or {} + active = report.get(route_id) if route_id else None + + states: dict[str, dict] = {} + missing: set[tuple[str, str]] = set() + health = getattr(backend, "health", None) + if health is not None: + snapshot = health.snapshot() + states = snapshot.get("targets", {}) + missing = {(m["target"], m["model"]) for m in snapshot.get("missing_models", [])} + + chain: list[dict] = [] + if route is not None: + for entry in route.entries: + state = states.get(entry.target.name, {}) + item = { + "model": entry.model, + "target": entry.target.name, + "host": entry.target.host, + "primary": entry is route.primary, + "state": "missing" if (entry.target.name, entry.model) in missing + else state.get("state", "ok"), + } + if state.get("reason"): + item["reason"] = state["reason"] + chain.append(item) + info["chain"] = chain + + if active is not None: + host = next( + (e["host"] for e in chain + if e["model"] == active["model"] and e["target"] == active["target"]), + "", + ) + info["active"] = { + "model": active["model"], + "target": active["target"], + "host": host, + "primary": bool(active.get("primary", True)), + "context_window": _effective_window(profile, backend), + } + return info + +def _recall_payload(recall: Any) -> dict | None: + """The recall as the client reads it — one shape everywhere.""" + if recall is None: + return None + return { + "id": recall.id, + "call_type": recall.call_type, + "trigger_at": recall.trigger_at.isoformat(), + "interval_seconds": recall.interval_seconds, + "internal_comment": recall.internal_comment, + "additional_context_message": recall.additional_context_message, + "status": recall.status, + "created_at": recall.created_at.isoformat(), + } + +def _require_share_owner(session, user: User) -> None: + """Only the owner (or an admin) may hand out a chat's history. + + Deliberately stricter than ``check_session_access``: a read-all permission + lets staff *read* other users' sessions, not distribute them. + """ + if not settings.navi_auth_enabled: + if session.user_id is not None: + raise HTTPException(status_code=403, detail="Access denied") + return + if session.user_id is not None and session.user_id == user.id: + return + if user.role == "admin": + return + raise HTTPException(status_code=403, detail="Access denied") diff --git a/navi/api/routes/sessions/detail.py b/navi/api/routes/sessions/detail.py new file mode 100644 index 0000000..694ccc6 --- /dev/null +++ b/navi/api/routes/sessions/detail.py @@ -0,0 +1,184 @@ +"""The session head and read-only debug views (meta, info, context, planning, content).""" + +from typing import Annotated, Any + +from fastapi import Depends, HTTPException + +from navi.api.deps import ( + get_backend_registry, + get_orchestrator, + get_profile_registry, + get_scheduler, + get_session_store, + require_admin, + require_user, +) +from navi.auth import User +from navi.auth.deps import check_session_access +from navi.config import settings +from navi.content_store import list_for_session +from navi.core import BackendRegistry, ProfileRegistry, SessionStore +from navi.core.scheduler import RecallScheduler + +from ._router import router # the shared /sessions router +from .common import ( + _backend_for, + _effective_window, + _llm_info, + _profile_of, + _recall_payload, +) + + +@router.get("/{session_id}/meta") +async def get_session_meta( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], + backends: Annotated[BackendRegistry, Depends(get_backend_registry)], + user: Annotated[User, Depends(require_user)], +) -> dict: + # Light session head without the message payload — the webclient loads the + # history separately via /{session_id}/messages/history in pages. + session = await store.get_meta(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + profile = _profile_of(profiles, session.profile_id) + backend = _backend_for(profile, backends) + return { + "session_id": session.id, + "profile_id": session.profile_id, + "name": session.name, + "message_count": session.db_message_count, + "context_token_count": session.context_token_count, + "max_context_tokens": _effective_window(profile, backend), + "archive_threshold": session.archive_threshold, + "created_at": session.created_at.isoformat(), + "last_active": session.last_active.isoformat(), + } + +@router.get("/{session_id}/info") +async def get_session_info( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], + backends: Annotated[BackendRegistry, Depends(get_backend_registry)], + orchestrator: Annotated[Any, Depends(get_orchestrator)], + scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Everything the session-info dialog shows, in one authenticated call. + + The model identity lives in the route engine, not in the session row, so + this joins the session head with the profile's route: the active entry + (model, target, host, window) and the health of every entry in the chain. + Reads in-process state only — the background probe keeps it current. + """ + session = await store.get_meta(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + + profile = _profile_of(profiles, session.profile_id) + backend = _backend_for(profile, backends) + + keep_recent = getattr(profile, "compression_keep_recent", None) if profile else None + summary_max = getattr(profile, "compression_max_tokens", None) if profile else None + + return { + "session_id": session.id, + "name": session.name, + "profile_id": session.profile_id, + "profile_name": getattr(profile, "name", None) if profile else None, + "created_at": session.created_at.isoformat(), + "last_active": session.last_active.isoformat(), + "message_count": session.db_message_count, + "context_token_count": session.context_token_count, + "max_context_tokens": _effective_window(profile, backend), + "archive_threshold": session.archive_threshold, + "compression": { + "enabled": settings.context_compression_enabled, + "threshold": settings.context_compression_threshold, + "keep_recent": keep_recent or settings.context_keep_recent, + "summary_max_tokens": summary_max or settings.context_summary_max_tokens, + }, + "llm": _llm_info(profile, backend), + "live": { + "running": orchestrator.is_running(session_id), + "clients": orchestrator.client_count(session_id), + }, + "recall": _recall_payload(await scheduler.get_pending_recall(session_id)), + } + +@router.get("/{session_id}") +async def get_session( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], + backends: Annotated[BackendRegistry, Depends(get_backend_registry)], + user: Annotated[User, Depends(require_user)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + profile = _profile_of(profiles, session.profile_id) + backend = _backend_for(profile, backends) + return { + "session_id": session.id, + "profile_id": session.profile_id, + "name": session.name, + "messages": [m.model_dump(mode='json', exclude_none=True) for m in session.messages], + "context_token_count": session.context_token_count, + "max_context_tokens": _effective_window(profile, backend), + "archive_threshold": session.archive_threshold, + "created_at": session.created_at.isoformat(), + "last_active": session.last_active.isoformat(), + } + +@router.get("/{session_id}/context") +async def get_session_context( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_admin)], +) -> dict: + """Return the LLM context (what the model actually sees) for debugging.""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + total_chars = sum(len(m.content or "") for m in session.context) + return { + "session_id": session.id, + "profile_id": session.profile_id, + "message_count": len(session.context), + "total_chars": total_chars, + "context": [m.model_dump(mode="json", exclude_none=True) for m in session.context], + } + +@router.get("/{session_id}/planning") +async def get_session_planning( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_admin)], +) -> dict: + """Return all planning phase debug logs for a session.""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + return {"session_id": session.id, "logs": session.planning_logs} + +@router.get("/{session_id}/content") +async def get_session_content( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], +) -> dict: + """Return published inline content records for this session. + + Public — no auth required. The session ID acts as an unguessable + capability token for accessing shared/published files. + """ + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + return {"session_id": session.id, "content": await list_for_session(session_id)} diff --git a/navi/api/routes/sessions/files.py b/navi/api/routes/sessions/files.py new file mode 100644 index 0000000..3045ffc --- /dev/null +++ b/navi/api/routes/sessions/files.py @@ -0,0 +1,140 @@ +"""Session file upload, listing and download.""" + +import asyncio +import mimetypes +from typing import Annotated + +from fastapi import Depends, HTTPException, UploadFile +from fastapi.responses import FileResponse + +from navi.api.deps import get_session_store, require_user +from navi.auth import User +from navi.auth.deps import check_session_access +from navi.config import settings +from navi.content_types import detect, is_inline_mime +from navi.core import SessionStore +from navi.session_files import ( + ensure_session_dir, + is_forbidden, + list_session_files, + safe_filename, + session_dir, +) + +from ._router import router # the shared /sessions router + + +@router.post("/{session_id}/files", status_code=201) +async def upload_file( + session_id: str, + file: UploadFile, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + + name = safe_filename(file.filename or "upload") + if is_forbidden(name): + raise HTTPException(status_code=400, detail="Executable files are not allowed") + + max_bytes = settings.session_files_max_size_mb * 1024 * 1024 + dest = ensure_session_dir(session_id) / name + + # Handle duplicate filenames + if dest.exists(): + stem, suffix = dest.stem, dest.suffix + i = 1 + while dest.exists(): + dest = dest.with_name(f"{stem}_{i}{suffix}") + i += 1 + + # Stream chunks directly to disk to avoid buffering the whole file in RAM + size = 0 + try: + with dest.open("wb") as f: + while True: + chunk = await file.read(1024 * 1024) # 1 MB at a time + if not chunk: + break + size += len(chunk) + if size > max_bytes: + raise HTTPException( + status_code=413, + detail=f"File exceeds {settings.session_files_max_size_mb} MB limit", + ) + await asyncio.to_thread(f.write, chunk) + except HTTPException: + raise + except Exception as exc: + await asyncio.to_thread(dest.unlink, True) + raise HTTPException(status_code=500, detail=f"Upload failed: {exc}") from exc + + return { + "name": dest.name, + "size": size, + "path": str(dest), + "content_type": file.content_type or "application/octet-stream", + } + +@router.get("/{session_id}/files") +async def list_session_files_endpoint( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Return all files and directories in the session's file directory (recursive, depth 10).""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.files.read_all") + + files = list_session_files(session_id) + return {"session_id": session_id, "files": files} + +@router.get("/{session_id}/files/{filename}") +async def download_file( + session_id: str, + filename: str, + store: Annotated[SessionStore, Depends(get_session_store)], + download: bool = False, +) -> FileResponse: + """Download a file from the session's file directory. + + Public — no auth required. The session ID acts as an unguessable + capability token for accessing shared/published files. + """ + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + + # Resolve and verify the file is within the session directory (no path traversal) + base = session_dir(session_id).resolve() + file_path = (base / filename).resolve() + try: + file_path.relative_to(base) + except ValueError: + raise HTTPException(status_code=403, detail="Access denied") + + if not file_path.exists() or not file_path.is_file(): + raise HTTPException(status_code=404, detail="File not found") + + # The session's own registry decides the type — the host's /etc/mime.types + # is the fallback for formats it does not cover (docx, archives), where a + # sensible MIME still makes the saved file openable. + _, content_type = detect(file_path.name, file_path) + if content_type == "application/octet-stream": + content_type = mimetypes.guess_type(file_path.name)[0] or content_type + # Video and audio belong in this list as much as images do: served as an + # attachment they are downloaded instead of played, even when the client + # asks the browser to open the URL rather than the in-page player. + inline = not download and is_inline_mime(content_type) + + return FileResponse( + path=file_path, + filename=filename, + media_type=content_type, + content_disposition_type="inline" if inline else "attachment", + ) diff --git a/navi/api/routes/sessions/listing.py b/navi/api/routes/sessions/listing.py new file mode 100644 index 0000000..a3ea83e --- /dev/null +++ b/navi/api/routes/sessions/listing.py @@ -0,0 +1,196 @@ +"""Creating and listing sessions.""" + +import asyncio +from datetime import datetime, timedelta, timezone +from typing import Annotated + +import structlog +from fastapi import Depends, HTTPException, Query + +from navi.api.deps import ( + get_memory_store, + get_profile_registry, + get_scheduler, + get_session_store, + require_user, +) +from navi.auth import User +from navi.config import settings +from navi.core import ProfileRegistry, SessionStore +from navi.core.scheduler import RecallScheduler +from navi.exceptions import ProfileNotFound +from navi.memory import MemoryStore +from navi.profiles.base import admin_only_blocked + +from .common import CreateSessionRequest, _session_admin, _session_owner + +log = structlog.get_logger() +from ._router import router # the shared /sessions router + + +@router.post("", status_code=201) +async def create_session( + payload: CreateSessionRequest, + store: Annotated[SessionStore, Depends(get_session_store)], + profiles: Annotated[ProfileRegistry, Depends(get_profile_registry)], + memory: Annotated[MemoryStore, Depends(get_memory_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + profile_id = payload.profile_id + if not profile_id: + profile_id = settings.navi_default_profile_id + if not profile_id: + raise HTTPException(status_code=400, detail="profile_id is required and no default profile is configured") + + try: + profile = profiles.get(profile_id) + except ProfileNotFound: + raise HTTPException(status_code=404, detail=f"Profile '{profile_id}' not found") + + if admin_only_blocked(profile, user.role): + raise HTTPException( + status_code=403, detail=f"Profile '{profile_id}' requires admin access" + ) + + session = await store.create(profile_id, user_id=_session_owner(user)) + + # Fire-and-forget: extract memory from any stale sessions (last active > 30 min ago) + asyncio.create_task(_process_stale_sessions(store, memory, user_id=_session_owner(user))) + + return { + "session_id": session.id, + "profile_id": session.profile_id, + "created_at": session.created_at.isoformat(), + } + + +async def _process_stale_sessions(store: SessionStore, memory: MemoryStore, user_id: str | None = None) -> None: + """Extract facts from sessions idle for >30 min that haven't been processed yet.""" + from navi.api.deps import get_backend_registry + from navi.config import settings + from navi.memory.extractor import extract_and_update + + cutoff = datetime.now(timezone.utc) - timedelta(minutes=30) + try: + sessions = await store.list_all(user_id=user_id, is_admin=False) + backend = get_backend_registry().get("ollama") + except Exception: + return + + for session in sessions: + if not session.messages: + continue + if session.user_id is None: + continue # skip legacy sessions — no multi-user memory for unowned sessions + if session.last_active >= cutoff: + continue # still active + extracted_at = await memory.get_extracted_at(session.id) + if extracted_at and extracted_at >= session.last_active.isoformat(): + continue # already up to date + try: + await extract_and_update(session, backend, settings.ollama_default_model, memory) + except Exception: + log.warning("memory.extraction_failed", session_id=session.id, exc_info=True) + + +@router.get("") +async def list_sessions( + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], + scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], + limit: Annotated[int | None, Query(ge=1, le=100)] = None, + offset: Annotated[int | None, Query(ge=0)] = None, + profile_id: str | None = None, + search: str | None = Query(None), + special: bool | None = Query(None), +) -> dict | list[dict]: + # special=None and special=False both mean the regular list; special=True + # exposes the service sessions (e.g. Synapse reactions), which the sidebar + # hides behind its toggle. + special_filter = False if special is None else special + is_admin = _session_admin(user) + + if limit is None and offset is None: + sessions = await store.list_all(user_id=_session_owner(user), is_admin=is_admin, special=special_filter) + if profile_id and not search: + sessions = [s for s in sessions if s.profile_id == profile_id] + pending_ids = await scheduler.get_pending_session_ids([s.id for s in sessions]) + return [_session_summary(s, search=search, has_pending_recall=s.id in pending_ids) for s in sessions] + + page_limit = limit or 30 + page_offset = offset or 0 + if search: + sessions = await store.search_list( + limit=page_limit + 1, + offset=page_offset, + user_id=_session_owner(user), + is_admin=is_admin, + search=search, + special=special_filter, + ) + else: + sessions = await store.list_page( + limit=page_limit + 1, + offset=page_offset, + profile_id=profile_id, + user_id=_session_owner(user), + is_admin=is_admin, + special=special_filter, + ) + items = sessions[:page_limit] + pending_ids = await scheduler.get_pending_session_ids([s.id for s in items]) + return { + "items": [_session_summary(s, search=search, has_pending_recall=s.id in pending_ids) for s in items], + "limit": page_limit, + "offset": page_offset, + "has_more": len(sessions) > page_limit, + "next_offset": page_offset + len(items), + } + + +def _session_summary(session, search: str | None = None, has_pending_recall: bool = False) -> dict: + summary = { + "session_id": session.id, + "profile_id": session.profile_id, + "name": session.name, + "message_count": len(session.messages), + "preview": _preview(session), + "pinned": session.pinned, + "special": session.special, + "has_pending_recall": has_pending_recall, + "created_at": session.created_at.isoformat(), + "last_active": session.last_active.isoformat(), + } + if search: + summary["match_indices"], summary["match_preview"] = _match_info(session, search) + return summary + + +def _preview(session) -> str: + """Return a short snippet from the last user or assistant message.""" + for msg in reversed(session.messages): + if msg.role in ("user", "assistant") and msg.content: + return msg.content[:60] + return "" + + +def _match_info(session, search: str) -> tuple[list[int], str]: + """Return message indices that contain the search term and a preview snippet.""" + q = search.lower() + indices = [] + preview = "" + for i, msg in enumerate(session.messages): + content = msg.content or "" + if q in content.lower(): + indices.append(i) + if not preview: + idx = content.lower().find(q) + start = max(0, idx - 30) + end = min(len(content), idx + len(search) + 30) + snippet = content[start:end] + if start > 0: + snippet = "…" + snippet + if end < len(content): + snippet = snippet + "…" + preview = snippet + return indices, preview diff --git a/navi/api/routes/sessions/manage.py b/navi/api/routes/sessions/manage.py new file mode 100644 index 0000000..7b44ca5 --- /dev/null +++ b/navi/api/routes/sessions/manage.py @@ -0,0 +1,229 @@ +"""Session lifecycle and runtime state: pin, delete, share, todos, terminals, tasks.""" + +from typing import Annotated + +import structlog +from fastapi import Depends, HTTPException + +from navi.api.deps import ( + get_kv_store, + get_session_store, + get_todo_store, + require_user, +) +from navi.auth import User +from navi.auth.deps import check_session_access +from navi.config import settings +from navi.core import SessionStore +from navi.session_files import delete_session_dir +from navi.store import KvStore +from navi.todo_store import PgTodoStore + +from .common import PinSessionRequest, _require_share_owner + +log = structlog.get_logger() +from ._router import router # the shared /sessions router + + +@router.get("/{session_id}/todos") +async def get_session_todos( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + kv: Annotated[KvStore, Depends(get_kv_store)], + todos: Annotated[PgTodoStore, Depends(get_todo_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Return the session's current todo list (the parent agent's plan). + + Reads the parent session's todo row via explicit (user_id, session_id) — no + ContextVars. Sub-agent todos live in isolated rows and are not exposed here + (Phase 3). Used by the TUI to seed the side panel on attach/switch. + """ + from navi.tools.todo import load_tasks_for + + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + tasks = await load_tasks_for(session_id, session.user_id, kv=kv, store=todos) + return {"session_id": session_id, "tasks": tasks} + +@router.get("/{session_id}/terminals") +async def list_session_terminals( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """List the session's active persistent terminals (background dev servers, + watchers, …). Seeds the TUI terminals modal + status-bar count on attach.""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + + from navi.api.deps import get_orchestrator + + orchestrator = get_orchestrator() + terminals: list[dict] = [] + if orchestrator and orchestrator._container and orchestrator._container.terminal_manager: + terminals = orchestrator._container.terminal_manager.list(session_id) + return {"session_id": session_id, "terminals": terminals} + + +@router.post("/{session_id}/terminals/{terminal_name}/close") +async def close_session_terminal( + session_id: str, + terminal_name: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Force-close one persistent terminal by name (from the TUI /terminals modal).""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + + from navi.api.deps import get_orchestrator + + orchestrator = get_orchestrator() + if not (orchestrator and orchestrator._container and orchestrator._container.terminal_manager): + return {"session_id": session_id, "terminal_name": terminal_name, "closed": False} + closed = await orchestrator._container.terminal_manager.close(session_id, terminal_name) + return {"session_id": session_id, "terminal_name": terminal_name, "closed": closed} + +@router.patch("/{session_id}/pin") +async def pin_session( + session_id: str, + body: PinSessionRequest, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + ok = await store.set_pinned(session_id, body.pinned) + if not ok: + raise HTTPException(status_code=404, detail="Session not found") + return {"session_id": session_id, "pinned": body.pinned} + +@router.get("/{session_id}/tasks") +async def list_session_tasks( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """List this session's background tasks (the TaskManager jobs). + + task_update events are delivered live over the WS and are not replayed on + reconnect, so the client fetches this snapshot to rebuild its + backgrounds tab after a reload / reconnection. + """ + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.read_all") + + from datetime import datetime + from datetime import timezone as _tz + + from navi.core.tasks import args_summary, get_task_manager + + items = [] + for j in get_task_manager().list(session_id): + items.append({ + "task_id": j.task_id, + "tool": j.tool, + "status": j.status, + "args_summary": args_summary(j.args), + "parent_tool_call_id": j.parent_tool_call_id or None, + "started_at": datetime.fromtimestamp(j.created_at, tz=_tz.utc).isoformat(), + "finished_at": ( + datetime.fromtimestamp(j.finished_at, tz=_tz.utc).isoformat() + if j.finished_at is not None else None + ), + "subagent_tokens": j.subagent_tokens, + "result_preview": j.preview() or None, + }) + return {"session_id": session_id, "tasks": items} + +@router.delete("/{session_id}", status_code=204) +async def delete_session( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> None: + import time as _time + + _t0 = _time.perf_counter() + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user, permission="navi.sessions.delete_all") + # Close any persistent terminals tied to this session before deleting + from navi.api.deps import get_orchestrator + orchestrator = get_orchestrator() + if orchestrator and orchestrator._container and orchestrator._container.terminal_manager: + await orchestrator._container.terminal_manager.close_all(session_id) + + # Cancel background tasks for this session: they outlive the run but not + # their session, and letting them finish would burn tokens for a session + # that no longer exists while their completion notes dangle in the KV + # store (suppress_note = True, set inside cancel_session_jobs). + from navi.core.task_notes import clear as clear_task_notes + from navi.core.tasks import get_task_manager + _t_get = _time.perf_counter() + get_task_manager().cancel_session_jobs(session_id) + # Pending notes from earlier completed jobs would never drain — the + # session is going away. + await clear_task_notes(session_id) + _t_tasks = _time.perf_counter() + + deleted = await store.delete(session_id) + _t_db = _time.perf_counter() + if not deleted: + raise HTTPException(status_code=404, detail="Session not found") + # Remove session file directory if it exists + await delete_session_dir(session_id) + _t_dir = _time.perf_counter() + log.info( + "sessions.delete_timing", + session_id=session_id, + lookup_terminals_ms=round((_t_get - _t0) * 1000, 1), + tasks_notes_ms=round((_t_tasks - _t_get) * 1000, 1), + db_delete_ms=round((_t_db - _t_tasks) * 1000, 1), + files_dir_ms=round((_t_dir - _t_db) * 1000, 1), + total_ms=round((_t_dir - _t0) * 1000, 1), + ) + +@router.post("/{session_id}/share") +async def share_session( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Issue (or re-issue) the session's public read-only link. + + The link is a capability: whoever holds it reads the display history and + nothing else — no input, no session list, no artifacts panel. It has no + expiry and no revoke switch by decision; deleting the session closes it. + One share per session, so a second call hands back the same link instead + of multiplying exposure. + """ + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + _require_share_owner(session, user) + + share = await store.get_or_create_share(session_id, session.user_id) + if share is None: + # Lazily-persisted session: no row yet, so there is nothing to point at. + raise HTTPException(status_code=409, detail="Session has no saved messages yet") + + token = share["token"] + log.info("sessions.shared", session_id=session_id, user_id=user.id) + return { + "token": token, + "url": f"{settings.public_url.rstrip('/')}/#share={token}", + "created_at": share["created_at"].isoformat(), + } diff --git a/navi/api/routes/sessions/messages.py b/navi/api/routes/sessions/messages.py new file mode 100644 index 0000000..47c7dbd --- /dev/null +++ b/navi/api/routes/sessions/messages.py @@ -0,0 +1,59 @@ +"""Paged display history: archived pages and the archive ∪ hot page.""" + +from typing import Annotated + +from fastapi import Depends, HTTPException, Query + +from navi.api.deps import get_session_store, require_user +from navi.auth import User +from navi.auth.deps import check_session_access +from navi.core import SessionStore + +from ._router import router # the shared /sessions router + + +@router.get("/{session_id}/messages/archive") +async def get_session_archived_messages( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], + before_seq: int | None = None, + limit: int = Query(50, ge=1, le=200), +) -> dict: + """Return older archived messages for scroll-up loading.""" + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + msgs = await store.get_archived_messages(session_id, before_seq=before_seq, limit=limit) + next_before_seq = msgs[0].sequence_number if msgs else None + return { + "items": [m.model_dump(mode="json", exclude_none=True) for m in msgs], + "has_more": len(msgs) == limit, + "next_before_seq": next_before_seq, + } + +@router.get("/{session_id}/messages/history") +async def get_session_history( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], + before_seq: int | None = None, + limit: int = Query(150, ge=1, le=200), +) -> dict: + """Paged full history (archive ∪ hot), newest page first. + + Without ``before_seq`` — the tail a client shows initially; with it — the + next older page for scroll-up loading. Returns oldest-first items so the + client can render/prepend them directly. + """ + session = await store.get_meta(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + msgs, has_more = await store.get_history_page(session_id, before_seq=before_seq, limit=limit) + return { + "items": [m.model_dump(mode="json", exclude_none=True) for m in msgs], + "has_more": has_more, + "next_before_seq": msgs[0].sequence_number if msgs else None, + } diff --git a/navi/api/routes/sessions/naming.py b/navi/api/routes/sessions/naming.py new file mode 100644 index 0000000..941b1a9 --- /dev/null +++ b/navi/api/routes/sessions/naming.py @@ -0,0 +1,101 @@ +"""Session titles: LLM generation and manual rename.""" + +from typing import Annotated + +from fastapi import Depends, HTTPException + +from navi.api.deps import get_backend_registry, get_session_store, require_user +from navi.auth import User +from navi.auth.deps import check_session_access +from navi.config import settings +from navi.core import BackendRegistry, SessionStore +from navi.core.name_generator import generate_session_name + +from ._router import router # the shared /sessions router +from .common import RenameSessionRequest + + +@router.post("/{session_id}/generate-name") +async def generate_name( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + backends: Annotated[BackendRegistry, Depends(get_backend_registry)], + user: Annotated[User, Depends(require_user)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + + # Only generate if no name yet + if session.name: + return {"name": session.name} + + user_messages = [ + m.content for m in session.messages + if m.role == "user" and m.content + ] + if not user_messages: + return {"name": None} + + # Short first user messages often carry no topic (greetings, one-liners) — + # that's why a share of sessions stayed unnamed. The agent's final replies + # are where the actual subject lands; fold them into the naming context. + assistant_messages = [ + m.content for m in session.messages + if m.role == "assistant" and m.content + ][-2:] + + backend = backends.get("ollama") + name = await generate_session_name( + user_messages, backend, settings.ollama_default_model, + assistant_messages=assistant_messages, + ) + + if name: + await store.set_name(session_id, name) + + return {"name": name} + +@router.patch("/{session_id}/name") +async def rename_session( + session_id: str, + body: RenameSessionRequest, + store: Annotated[SessionStore, Depends(get_session_store)], + backends: Annotated[BackendRegistry, Depends(get_backend_registry)], + user: Annotated[User, Depends(require_user)], +) -> dict: + """Set the session title manually, or regenerate it on an empty edit. + + A whitespace-only name falls back to LLM naming (the same context as + generate-name: user messages plus the agent's final replies). + """ + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + + name = body.name.strip()[:200] + if not name: + user_messages = [ + m.content for m in session.messages + if m.role == "user" and m.content + ] + assistant_messages = [ + m.content for m in session.messages + if m.role == "assistant" and m.content + ][-2:] + if not user_messages: + return {"name": None} + backend = backends.get("ollama") + name = await generate_session_name( + user_messages, backend, settings.ollama_default_model, + assistant_messages=assistant_messages, + ) + if not name: + return {"name": None} + + ok = await store.set_name(session_id, name) + if not ok: + raise HTTPException(status_code=404, detail="Session not found") + return {"session_id": session_id, "name": name} diff --git a/navi/api/routes/sessions/recall.py b/navi/api/routes/sessions/recall.py new file mode 100644 index 0000000..b58c2bb --- /dev/null +++ b/navi/api/routes/sessions/recall.py @@ -0,0 +1,70 @@ +"""Scheduled recalls attached to a session.""" + +from typing import Annotated + +from fastapi import Depends, HTTPException + +from navi.api.deps import get_scheduler, get_session_store, require_user +from navi.auth import User +from navi.auth.deps import check_session_access +from navi.core import SessionStore +from navi.core.scheduler import RecallScheduler + +from ._router import router # the shared /sessions router +from .common import _recall_payload + + +@router.get("/{session_id}/recall") +async def get_session_recall( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], + scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + return {"recall": _recall_payload(await scheduler.get_pending_recall(session_id))} + + +@router.delete("/{session_id}/recall") +async def cancel_session_recall( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], + scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + ok = await scheduler.cancel_recall(session_id) + if ok: + from navi.core.event_bus import get_event_bus + from navi.core.events import RecallUpdate + await get_event_bus().publish( + RecallUpdate(session_id=session_id, status="cancelled", action="cancelled") + ) + return {"ok": ok} + + +@router.post("/{session_id}/recall/skip") +async def skip_session_recall( + session_id: str, + store: Annotated[SessionStore, Depends(get_session_store)], + user: Annotated[User, Depends(require_user)], + scheduler: Annotated[RecallScheduler, Depends(get_scheduler)], +) -> dict: + session = await store.get(session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session not found") + check_session_access(session, user) + ok = await scheduler.skip_next_recall(session_id) + if ok: + from navi.core.event_bus import get_event_bus + from navi.core.events import RecallUpdate + await get_event_bus().publish( + RecallUpdate(session_id=session_id, status="pending", action="rescheduled") + ) + return {"ok": ok} diff --git a/tests/integration/test_auth_disabled.py b/tests/integration/test_auth_disabled.py index cb8f1b4..1ef3b60 100644 --- a/tests/integration/test_auth_disabled.py +++ b/tests/integration/test_auth_disabled.py @@ -23,7 +23,10 @@ monkeypatch.setattr("navi.config.settings", new_settings) monkeypatch.setattr("navi.auth.deps.settings", new_settings) monkeypatch.setattr("navi.api.routes.auth.settings", new_settings) - monkeypatch.setattr("navi.api.routes.sessions.settings", new_settings) + from navi.api.routes.sessions import common, detail, files, listing, manage, naming + + for _sessions_mod in (common, detail, files, listing, manage, naming): + monkeypatch.setattr(_sessions_mod, "settings", new_settings) store = InMemorySessionStore() profiles = make_profile_registry() @@ -157,7 +160,7 @@ async def test_create_session_uses_default_profile(self, monkeypatch, no_auth_client): """When navi_default_profile_id is set, POST /sessions with no body uses it.""" import navi.config as _config - from navi.api.routes import sessions as sessions_mod + from navi.api.routes.sessions import common, detail, files, listing, manage, naming new_settings = _config.Settings( database_url="postgresql://fake", @@ -167,7 +170,8 @@ monkeypatch.setattr("navi.config.settings", new_settings) monkeypatch.setattr("navi.auth.deps.settings", new_settings) monkeypatch.setattr("navi.api.routes.auth.settings", new_settings) - monkeypatch.setattr(sessions_mod, "settings", new_settings) + for _sessions_mod in (common, detail, files, listing, manage, naming): + monkeypatch.setattr(_sessions_mod, "settings", new_settings) client, store = no_auth_client response = client.post("/sessions", json={}) diff --git a/tests/unit/api/test_session_files.py b/tests/unit/api/test_session_files.py index 07b8c77..fa5c9ae 100644 --- a/tests/unit/api/test_session_files.py +++ b/tests/unit/api/test_session_files.py @@ -28,12 +28,14 @@ async def _to_thread(func, *args, **kwargs): return func(*args, **kwargs) - monkeypatch.setattr(sessions_mod.asyncio, "to_thread", _to_thread) + from navi.api.routes.sessions import files as files_mod + + monkeypatch.setattr(files_mod.asyncio, "to_thread", _to_thread) _test_settings = Settings( session_files_dir=str(tmp_path / "sessions"), session_files_max_size_mb=1, ) - monkeypatch.setattr(sessions_mod, "settings", _test_settings) + monkeypatch.setattr(files_mod, "settings", _test_settings) monkeypatch.setattr(session_files_mod, "settings", _test_settings) @@ -191,7 +193,9 @@ assert sid == session_id return [{"filename": "logo.svg"}] - monkeypatch.setattr(sessions_mod, "list_for_session", fake_list_for_session) + from navi.api.routes.sessions import detail as detail_mod + + monkeypatch.setattr(detail_mod, "list_for_session", fake_list_for_session) result = await sessions_mod.get_session_content(session_id, store) diff --git a/tests/unit/api/test_shared_session.py b/tests/unit/api/test_shared_session.py index 1e2d3f6..27af026 100644 --- a/tests/unit/api/test_shared_session.py +++ b/tests/unit/api/test_shared_session.py @@ -120,11 +120,12 @@ class TestShareSession: def _settings(self, monkeypatch, *, auth: bool = True): - monkeypatch.setattr( - sessions_mod, - "settings", - SimpleNamespace(public_url="https://navi.example", navi_auth_enabled=auth), - ) + from navi.api.routes.sessions import common as common_mod + from navi.api.routes.sessions import manage as manage_mod + + fake = SimpleNamespace(public_url="https://navi.example", navi_auth_enabled=auth) + monkeypatch.setattr(manage_mod, "settings", fake) + monkeypatch.setattr(common_mod, "settings", fake) async def test_the_owner_gets_a_link_to_the_read_only_page(self, monkeypatch): self._settings(monkeypatch)