From 02310511f457a347a9d854cc19814148024d7e7f Mon Sep 17 00:00:00 2001 From: kowshikdev Date: Thu, 23 Jul 2026 14:59:35 +0530 Subject: [PATCH 1/3] fix: verify and upgrade the stack, real Graph API data, generic gateway, Redis checkpointing Closes #1, #2, #4, #5. ## Verify the system runs end-to-end (#1) backend/app.py -- the actual FastAPI entrypoint -- never imported successfully. backend/routes_ai.py does `from app.smart_chat import SmartChatAgent`, and that module was never committed anywhere in this repository. One missing class took the entire app down at import time, over three endpoints (/assistant/start, /assistant/chat) out of several dozen. The import is now optional; only those three endpoints are affected and they answer 501 naming exactly what's missing, rather than the whole app refusing to boot. backend.app now imports cleanly and registers 74 routes -- first time, per the previous CLAUDE.md's own admission that nothing had been shown to run. SmartChatAgent itself is still not implemented (tracked separately -- no spec exists to build it from; that's real feature work, not something to reverse-engineer while fixing an import). ## Dependencies (#2) Checked empirically rather than assumed "expect real migration work, not a version bump": - langgraph 0.2.45 -> 1.x: the actual API surface this codebase touches (StateGraph, END, compile(checkpointer=...), invoke with thread_id) is UNCHANGED across that jump. Verified by constructing and invoking a graph against the installed 1.x. - litellm==1.42.6 was dead weight -- nothing in this codebase has ever imported the litellm package. governance/litellm_gateway.py always talked to an OpenAI-API-compatible endpoint directly via the openai SDK. Removed. - langchain-core pin removed -- nothing here imports langchain_core directly; arrives transitively via langgraph. - backend/requirements.txt allowed pydantic>=1.10, which would install a v1 pydantic and break every v2-only schema in backend/models.py and agents/schemas.py outright. Tightened to >=2.7,<3 in both requirement files. ## Mail and calendar: real Microsoft Graph data, not mock JSON repos/data_repo.py's inbox()/meetings()/get_transcript()/create_meeting() are backed by live Graph calls (repos/graph_client.py, auth via repos/graph_auth.py) -- not JSON fixtures. Device-code flow supports both personal Microsoft accounts and work/school tenant accounts through the same app registration (tenant=common). DataRepo stays the single class every one of the 31 existing call sites already constructs (DataRepo()) -- Graph-backing is folded directly into it rather than adding a wrapper class that would need touching every call site. Personal Microsoft accounts cannot access meeting transcripts at all (OnlineMeetingTranscript.Read.All isn't grantable to personal accounts under any consent). get_transcript() catches the resulting 403/401, logs why clearly, and returns "" -- meeting_agent.py's EXISTING fallback (a summary from title + attendees) takes over from there, unchanged. This needed zero changes to meeting_agent.py. tasks/followups/eod/weekly/mom_entries/users stay JSON-backed deliberately -- Opspilot's own generated output (things the agents produce by processing real mail/calendar), not something Graph provides as input. add_email_to_inbox (a "demo sender portal" that injected fake mail into a local JSON store) was removed outright, not migrated -- it fabricated content, which is exactly what this integration exists to stop doing once the input is a real mailbox. See docs/graph_setup.md for the one-time Azure AD app-registration steps (interactive; only a human can do this) and scripts/graph_login.py for the device-code login itself. ## LLM gateway: generic, and no longer fabricates responses governance/litellm_gateway.py / governance/gateway.py had THREE separate layers of fabricated fallback -- PolicyGateway's own simulate, EnhancedLiteLLMGateway's simulate, and PolicyGateway re-simulating if the gateway raised -- all returning literal placeholder content ("decisions": ["Decision 1", "Decision 2"]) when unconfigured, indistinguishable from a real analysis to whatever called it. All three are gone. Config is now generic: LLM_BASE_URL / LLM_API_KEY / LLM_MODEL_ID / LLM_EMBEDDING_MODEL_ID, not tied to one proxy host or deployment name. Missing config raises LLMNotConfiguredError at construction, not on first call. A real API failure after retries raises; it does not return a canned string that reads like an answer. ## Checkpointing: Redis-backed, not in-process (#5 groundwork) Every graph that persisted state used MemorySaver -- in-process only, so a restart silently discarded every in-flight workflow. orchestration/checkpointer.py is the single shared factory, backed by langgraph-checkpoint-redis. A missing Redis is a loud CheckpointerUnavailable, not a silent fall-through to the exact in-memory problem this replaces. chat_workflow.py and followup_reporting_subgraphs.py imported MemorySaver but never used it -- dead imports, removed. ## Two UI stacks (#4) Resolved to one real stack, not a decision to make: streamlit and nicegui were pinned in requirements.txt but neither is imported anywhere in the codebase (`grep -rn "import streamlit\|import nicegui"` across the whole tree returns nothing). Next.js (frontend/, 9 pages) is the only real UI. Removed the two unused packages. ## CI (#5) .github/workflows/ci.yml -- runs the real pytest suite plus a byte-compile check. No live Graph or Redis credentials needed: Graph failure paths are tested via respx-stubbed HTTP, not live calls, and the one Redis-dependent test skips itself when no Redis is reachable. ## Tests tests/comprehensive_quality_tests.py looked like a test suite but isn't one pytest can run -- it's a live-server smoke script needing a running backend on :8002 and a configured LLM judge. New, actually pytest-discoverable: test_graph_client.py (HTTP boundary stubbed with respx), test_data_repo.py, test_gateway.py, test_checkpointer.py, test_app_boots.py. 48 pass, 2 skip (Redis-required cases). Co-Authored-By: Claude Opus 4.8 --- .env.example | 26 +- .github/workflows/ci.yml | 44 ++ .gitignore | 11 + CLAUDE.md | 219 +++++++- README.md | 24 +- backend/.env.example | 15 +- backend/requirements.txt | 4 +- backend/routes_ai.py | 36 +- config/settings.py | 56 +- data/.gitkeep | 0 data/calendar/.gitkeep | 0 data/emails/.gitkeep | 0 data/governance/.gitkeep | 0 data/nudges/.gitkeep | 0 data/reporting/.gitkeep | 0 data/tasks/.gitkeep | 0 data/wellness/.gitkeep | 0 docs/graph_setup.md | 106 ++++ governance/gateway.py | 375 ++------------ governance/litellm_gateway.py | 475 ++++++----------- orchestration/autonomous_graph.py | 4 +- orchestration/chat_workflow.py | 1 - orchestration/checkpointer.py | 78 +++ orchestration/followup_reporting_subgraphs.py | 1 - orchestration/meeting_subgraph.py | 4 +- orchestration/super_graph.py | 4 +- orchestration/task_subgraph.py | 4 +- orchestration/wellness_subgraph.py | 4 +- pytest.ini | 4 + repos/data_repo.py | 489 ++++++++++++------ repos/graph_auth.py | 164 ++++++ repos/graph_client.py | 210 ++++++++ requirements.txt | 41 +- scripts/graph_login.py | 65 +++ tests/comprehensive_quality_tests.py | 8 +- tests/conftest.py | 14 + tests/test_app_boots.py | 81 +++ tests/test_checkpointer.py | 77 +++ tests/test_data_repo.py | 176 +++++++ tests/test_gateway.py | 92 ++++ tests/test_graph_client.py | 178 +++++++ 41 files changed, 2190 insertions(+), 900 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 data/.gitkeep create mode 100644 data/calendar/.gitkeep create mode 100644 data/emails/.gitkeep create mode 100644 data/governance/.gitkeep create mode 100644 data/nudges/.gitkeep create mode 100644 data/reporting/.gitkeep create mode 100644 data/tasks/.gitkeep create mode 100644 data/wellness/.gitkeep create mode 100644 docs/graph_setup.md create mode 100644 orchestration/checkpointer.py create mode 100644 pytest.ini create mode 100644 repos/graph_auth.py create mode 100644 repos/graph_client.py create mode 100644 scripts/graph_login.py create mode 100644 tests/conftest.py create mode 100644 tests/test_app_boots.py create mode 100644 tests/test_checkpointer.py create mode 100644 tests/test_data_repo.py create mode 100644 tests/test_gateway.py create mode 100644 tests/test_graph_client.py diff --git a/.env.example b/.env.example index 11e82b7..ec637dc 100644 --- a/.env.example +++ b/.env.example @@ -1,14 +1,26 @@ -# Azure OpenAI via LiteLLM -AZURE_OPENAI_API_KEY="sk-vPq51du3iqjYS9SWyZV--w" -AZURE_OPENAI_API_BASE=https://smartops-qa04.eastus.cloudapp.azure.com/paas/smartgenie/litellm -# Model deployments you created in Azure OpenAI -AZURE_OPENAI_GPT4O_MINI_DEPLOYMENT=azure/sc-rnd-gpt-4o-mini-01 -AZURE_OPENAI_EMBEDDING_DEPLOYMENT=azure/ai-rnd-text-embedding-ada-002 +# LLM access. Generic OpenAI-API-compatible config -- point this at Azure +# OpenAI, a self-hosted LiteLLM proxy, OpenAI directly, or anything else that +# speaks the OpenAI chat-completions wire format. Not tied to any specific +# deployment name or host. +LLM_BASE_URL=https://your-endpoint.example.com/v1 +LLM_API_KEY=your-api-key-here +LLM_MODEL_ID=your-chat-model-deployment-name +LLM_EMBEDDING_MODEL_ID=your-embedding-model-deployment-name + +# Microsoft Graph (mail + calendar). client_id has no default -- it must be +# an app registration you create yourself. See docs/graph_setup.md. +# tenant=common allows both personal Microsoft accounts and work/school +# tenant accounts to sign in through the same registration. +MS_GRAPH_CLIENT_ID=your-app-registration-client-id +MS_GRAPH_TENANT=common # Runtime ENV=dev -DATA_DIR=./mock_data_json +# Opspilot's own generated state (tasks, follow-ups, reports, audit/usage +# logs, email-processing ledger) -- NOT mail/calendar content, which is live +# from Microsoft Graph and never persisted here. +DATA_DIR=./data POLICY_FILE=./governance/policies.json # Governance overrides (optional) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..0c5d8c6 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,44 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +jobs: + test: + name: Tests + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install -r backend/requirements.txt + pip install pytest pytest-asyncio respx + + # No live Graph or Redis credentials exist in CI, and none are needed: + # Graph-dependent code paths raise a clear GraphAuthError when + # unconfigured (tested directly), the Graph HTTP boundary itself is + # stubbed with respx, and the one test needing a reachable Redis skips + # itself when none is found on the default port. + - name: Run tests + run: pytest tests/ -q + + lint: + name: Syntax check + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Byte-compile all sources + run: python -m compileall -q agents orchestration backend governance memory repos config scripts tests diff --git a/.gitignore b/.gitignore index 5d8e498..c3714f0 100644 --- a/.gitignore +++ b/.gitignore @@ -32,6 +32,17 @@ Thumbs.db .env.local .env.*.local +# Microsoft Graph token cache -- contains a live refresh token, never commit +.graph_token_cache.json + +# Opspilot's own generated runtime state (tasks, follow-ups, reports, audit/ +# usage logs, email-processing ledger). Per-deployment data, not source -- +# same treatment as chroma_db/ and episodes/ below: directory structure is +# tracked via .gitkeep, generated content is not. +data/*.json +data/*/*.json +!data/**/.gitkeep + # Caching & Logs .llm_cache/ *.log diff --git a/CLAUDE.md b/CLAUDE.md index bda0fcc..fafa4fe 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,21 +10,43 @@ a governance layer, and cost tracking. The governance + cost-tracking layer is the differentiator. Most multi-agent demos skip exactly that part. +Mail and calendar are **live Microsoft Graph data**, not mock JSON. See +"Microsoft Graph integration" below before touching `repos/`. + ## Stack -Python · LangGraph · LiteLLM · ChromaDB (vector memory) · FastAPI · Next.js · -Streamlit / NiceGUI +Python · LangGraph 1.x (Redis-backed checkpointing) · Microsoft Graph (mail, +calendar) · Generic OpenAI-API-compatible LLM gateway · ChromaDB (vector +memory) · FastAPI · Next.js + +`streamlit` and `nicegui` were pinned in requirements.txt as a supposed +"second UI stack" but were never imported anywhere in the codebase -- +removed. Next.js (`frontend/`, 9 pages) is the only real UI and is required. ## Layout - `agents/` — the six specialized agents (11 modules) -- `orchestration/` — LangGraph workflows and routing (11 modules) +- `orchestration/` — LangGraph workflows and routing (12 modules, including + `checkpointer.py`, the shared Redis checkpointer factory) - `backend/` — FastAPI server and routes (21 modules) -- `frontend/` — Next.js dashboard +- `frontend/` — Next.js dashboard. **Has its own static mock data** + (`frontend/public/data/*.json`) and its own "uses mock data as fallback" + behavior, untouched by this pass — see "What's still open" below. - `memory/` — vector store + episodic memory -- `governance/` — policy enforcement, audit logging, cost management -- `repos/` — data access layer -- `chroma_db/`, `episodes/` — persisted state, committed to the repo +- `governance/` — policy enforcement, audit logging, cost management, the LLM + gateway +- `repos/` — `data_repo.py` (mail/calendar via Graph + Opspilot's own + generated-state JSON), `graph_auth.py`, `graph_client.py` +- `data/` — Opspilot's own generated state (tasks, follow-ups, reports, + audit/usage logs, email-processing ledger). Directory structure is tracked + via `.gitkeep`; content is gitignored, same treatment as `chroma_db/` and + `episodes/` below. **This is not mock data** — it's the app's own output, + starting empty. +- `chroma_db/`, `episodes/` — persisted state, `.gitkeep`-tracked +- `docs/graph_setup.md` — Azure Portal app-registration steps (interactive; + only you can do this) +- `scripts/graph_login.py` — run this once to complete the Graph device-code + login - `ARCHITECTURE.md`, `CODEBASE_INDEX.md`, `super_graph.md` — good existing docs; read `ARCHITECTURE.md` first - `print_graph.py` — renders the LangGraph topology @@ -34,24 +56,169 @@ Streamlit / NiceGUI ```bash cp .env.example .env pip install -r requirements.txt + +# One-time: create an Azure AD app registration (docs/graph_setup.md), +# set MS_GRAPH_CLIENT_ID, then: +python -m scripts.graph_login + +# Needs a reachable Redis (checkpointing) and LLM_BASE_URL/LLM_API_KEY/ +# LLM_MODEL_ID (any OpenAI-API-compatible endpoint) to actually run agents. ``` -## Known issues - -- **Unverified.** 13,747 lines of Python landed in 3 commits on a single day - (2026-02-05), with one test file. Nothing here has been shown to run - end-to-end. Establishing that is the first task, before any feature work. -- **Deps are ~2 years stale and this is the blocker:** `langgraph==0.2.45`, - `litellm==1.42.6`, `langchain-core==0.3.20`, `streamlit==1.40.1`. LangGraph's - API moved substantially after 0.2.x — expect real migration work, not a - version bump. -- No CI, no LICENSE. -- Two UI stacks present (Next.js frontend *and* Streamlit/NiceGUI deps). Pick one. -- `chroma_db/` and `episodes/` hold runtime state but only `.gitkeep` is tracked, - so the directories exist without the data. That part is set up correctly. - -## If you're scaling this - -Order: prove it runs → upgrade the LangGraph/LiteLLM stack → per-agent evals → -drop the second UI → deploy a demo. The architecture is sound; the risk is -entirely in whether it executes. +## ✅ Deps upgraded and verified — FIXED + +The old CLAUDE.md said "expect real migration work, not a version bump." That +turned out to be mostly wrong, and it was worth checking rather than assuming: + +- **`langgraph` 0.2.45 → 1.x**: the actual API surface this codebase touches + (`StateGraph`, `END`, `add_node`/`add_edge`/`set_entry_point`, + `compile(checkpointer=...)`, `invoke(state, {"configurable": {"thread_id": ...}})`) + is **unchanged** across that jump. Verified by constructing and invoking a + graph against the installed 1.x, not by reading a migration guide. +- **`litellm==1.42.6` was dead weight.** Nothing in this codebase has ever + imported the `litellm` package — `governance/litellm_gateway.py` always + talked to an OpenAI-API-compatible endpoint directly via the `openai` SDK. + Removed rather than upgraded. +- **`langchain-core` pin removed** — nothing here imports `langchain_core` + directly; it arrives transitively via `langgraph`. +- **`pydantic>=1.10` in `backend/requirements.txt` was a real landmine**: every + schema in `backend/models.py` / `agents/schemas.py` is v2-only syntax. Tightened + to `>=2.7,<3` in both requirement files. + +## ✅ Microsoft Graph integration — mail and calendar are real, not mock + +`repos/data_repo.py`'s `inbox()`, `meetings()`, `get_transcript()`, and +`create_meeting()` are backed by live Microsoft Graph calls +(`repos/graph_client.py`, auth via `repos/graph_auth.py`), not JSON fixtures. +**There is no local mail/calendar mock data anywhere in the backend on +purpose** — see `docs/graph_setup.md` for the one-time app-registration setup +this requires, and `scripts/graph_login.py` for the login itself. + +Design points worth knowing before touching this: + +- **`DataRepo` stays the single class every call site already constructs** + (`DataRepo()`, 31 call sites). Graph-backing is folded directly into it + rather than adding a wrapper class that would need touching every + construction site. +- **Field mapping lives in `_map_message`/`_map_event`** in `data_repo.py`, + translating Graph's field names into the shapes `backend/models.py` and the + agents already expect. Nothing downstream had to change. +- **`transcript_file` is reused, not repurposed.** It has always meant "the + identifier `get_transcript()` needs." Under Graph that identifier is a Teams + join URL, not a filename — the call site + (`get_transcript(mtg.get("transcript_file"))`) is unchanged. +- **Personal Microsoft accounts cannot access meeting transcripts at all** — + `OnlineMeetingTranscript.Read.All` isn't grantable to personal accounts under + any consent; on a work/school tenant it needs admin consent. `get_transcript` + catches the resulting 403/401, logs *why* clearly, and returns `""`. + `meeting_agent.py`'s existing fallback (a summary from title + attendees) + takes over from there — this needed **zero changes** to `meeting_agent.py`. +- **`tasks`, `followups`, `eod`/`weekly`, `mom_entries`, `users` stay + JSON-backed.** These are Opspilot's own generated output (things the agents + produce by processing real mail/calendar), not something Graph provides as + input. Don't try to "migrate" these to Graph — there's nothing on the Graph + side to migrate them to. +- **Email processing state (`processed`/`agent_actions`/`agent_category`) is a + local ledger keyed by `email_id`** (`data/emails/email_state.json`), merged + into `inbox()`'s output. Graph has no concept of "has our agent triaged + this," so that state has to live somewhere, and it must never be confused + with mail content itself. +- **`create_meeting()` really schedules a calendar event** (`POST + /me/events`) — used by the `schedule_meeting` approved-action path in + `governance/approval.py`. It no longer appends to a local fake meetings list. +- **`add_email_to_inbox`** (a "demo sender portal" feature that injected fake + mail into a local JSON store) **was removed outright**, not migrated — it + fabricated content, which is exactly what this integration exists to stop + doing once the input is a real mailbox. + +Sync/async: Graph calls are async (`httpx.AsyncClient`); every existing call +site (`inbox()`, `meetings()`, etc.) is synchronous. `_run_sync()` bridges +this — safe from both a plain script and from inside a running event loop +(FastAPI), at the cost of a thread hop in the nested case. Migrating the 31 +call sites to native async is future work, not done here. + +## ✅ LLM gateway — generic, and no longer fabricates responses + +`governance/litellm_gateway.py` / `governance/gateway.py` used to have +**three separate layers of fabricated fallback** — `PolicyGateway`'s own +simulate, `EnhancedLiteLLMGateway`'s simulate, and `PolicyGateway` +re-simulating if the gateway raised — all returning literal placeholder +content (`"decisions": ["Decision 1", "Decision 2"]`) when unconfigured, +indistinguishable from a real analysis to whatever called it. All three are +gone. + +- Config is now generic: `LLM_BASE_URL` / `LLM_API_KEY` / `LLM_MODEL_ID` / + `LLM_EMBEDDING_MODEL_ID`, not tied to any specific proxy host or deployment + name (`SETTINGS["llm"]`, was `SETTINGS["models"]` with Azure-specific keys). +- **Fails at construction, not first call.** Missing config raises + `LLMNotConfiguredError` immediately — a misconfigured gateway should fail + when it's built, not three steps into a workflow. +- A real API failure after retries **raises**; it does not return a canned + string that reads like an answer. + +## ✅ Checkpointing — Redis-backed, not in-process + +Every graph that persisted state used `MemorySaver` — in-process only, so a +restart silently discarded every in-flight workflow. `orchestration/checkpointer.py` +is now the single shared factory (`get_checkpointer()`), backed by +`langgraph-checkpoint-redis`. Requires a reachable Redis +(`docker run -p 6379:6379 redis` for local dev); a missing one is a loud +`CheckpointerUnavailable`, not a silent fall-through to the exact in-memory +problem this replaces. + +`chat_workflow.py` and `followup_reporting_subgraphs.py` imported +`MemorySaver` but never actually used it — those were dead imports, now +removed rather than migrated. + +## ✅ The app has been shown to boot — first time, per the old CLAUDE.md + +`backend/app.py` failed to import at all: `backend/routes_ai.py` imports +`from app.smart_chat import SmartChatAgent`, and **that module was never +committed anywhere in this repository**. One missing class took down the +entire FastAPI app over three endpoints (`/assistant/start`, `/assistant/chat`, +the third doesn't touch it) out of several dozen. + +The import is now optional; only those three endpoints are affected, and they +answer **501** naming exactly what's missing. **`SmartChatAgent` itself is +still not implemented** — there's no spec anywhere to reconstruct it from, and +building an undefined conversational-agent class wasn't part of this pass. +`backend/app.py` now imports cleanly and registers 74 routes. + +## ✅ Real tests + CI — first time + +`tests/comprehensive_quality_tests.py` looked like a test suite but isn't one +pytest can run: it's a live-server smoke script (`python tests/comprehensive_quality_tests.py`) +that hits a running backend on `localhost:8002` and uses an LLM as a judge. It +still needs a running server and a configured LLM to do anything. + +New, actually pytest-discoverable: `test_graph_client.py` (HTTP boundary +stubbed with `respx` — test-boundary mocking, not product mock data), +`test_data_repo.py`, `test_gateway.py`, `test_checkpointer.py`, +`test_app_boots.py`. **48 pass, 2 skip** (Redis-required cases, skip +themselves when no Redis is reachable on 6379). `.github/workflows/ci.yml` +runs the suite on every push/PR — no live Graph or Redis credentials needed; +Graph failure paths are tested via stubs, not live calls. + +## What's still open + +- **`SmartChatAgent` is unimplemented.** Building it is a real feature-design + task (no spec exists), not something to reverse-engineer. +- **The frontend has its own static mock data** + (`frontend/public/data/*.json`) and its own "mock data as fallback" comment + in `frontend/.env.example` — untouched by this pass. Fixing that is Next.js/ + TypeScript work, a separate chunk from the backend changes here. +- **deepagents migration** — discussed, not started. The six hand-rolled + ReAct agents (`agents/react_agent.py`) are a strong candidate to replace + with deepagents' subagent harness, now that the underlying LangGraph/Redis/ + Graph-API/LLM-gateway foundation is solid. That's the natural next phase. + +Resolved, not open: LICENSE already exists (MIT, root of repo) -- the earlier +note that it was missing was wrong. "Two UI stacks" resolved to one real +stack (Next.js) plus two unused pip packages, not a decision to make. + +## If you're scaling this further + +The foundation (deps, real data, gateway, checkpointing, boot verification, +tests, CI) is done. Order from here: deepagents migration for the six agents → +`SmartChatAgent` (if the conversational feature is still wanted) → drop the +second UI → frontend off its own static mock data → LICENSE. diff --git a/README.md b/README.md index 486b244..95c7eeb 100644 --- a/README.md +++ b/README.md @@ -214,13 +214,23 @@ Use `.env.example` as a template for your environment setup. Copy `.env.example` to `.env` and configure: ```bash -# LLM Configuration -AZURE_OPENAI_KEY=... -AZURE_OPENAI_ENDPOINT=... -AZURE_OPENAI_MODEL=... - -# Data Configuration -DATA_DIR=./data/mock_data_json +# LLM Configuration -- generic OpenAI-API-compatible; not tied to Azure +# specifically. See .env.example for the full set. +LLM_BASE_URL=... +LLM_API_KEY=... +LLM_MODEL_ID=... + +# Microsoft Graph (mail + calendar are live data, not mock JSON). +# See docs/graph_setup.md before setting these. +MS_GRAPH_CLIENT_ID=... +MS_GRAPH_TENANT=common + +# Data Configuration -- Opspilot's own generated state (tasks, follow-ups, +# reports, audit/usage logs). Mail/calendar are never stored here. +DATA_DIR=./data + +# Checkpointing -- requires a reachable Redis +REDIS_URL=redis://localhost:6379/0 # Server Configuration API_PORT=8002 diff --git a/backend/.env.example b/backend/.env.example index 25b280a..e636f9a 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -1,11 +1,16 @@ # Example .env for backend ENV=dev BACKEND_API_KEY=dev-unprotected -AZURE_OPENAI_API_KEY= -AZURE_OPENAI_API_BASE= -AZURE_OPENAI_GPT4O_MINI_DEPLOYMENT= -AZURE_OPENAI_EMBEDDING_DEPLOYMENT= -DATA_DIR=./data/mock_data_json + +# See ../.env.example for the full set of LLM / Graph variables and comments. +LLM_BASE_URL= +LLM_API_KEY= +LLM_MODEL_ID= +LLM_EMBEDDING_MODEL_ID= +MS_GRAPH_CLIENT_ID= +MS_GRAPH_TENANT=common + +DATA_DIR=./data POLICY_FILE=./governance/policies.json DAILY_BUDGET_USD=50 USE_CHROMADB=0 diff --git a/backend/requirements.txt b/backend/requirements.txt index ab7cf66..f32ed51 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -1,6 +1,8 @@ fastapi>=0.95.0 uvicorn[standard]>=0.22.0 -pydantic>=1.10 +# >=1.10 allowed a pydantic v1 install, which would break outright: every +# schema in backend/models.py and agents/schemas.py is v2-only syntax. +pydantic>=2.7,<3 anyio sse-starlette>=1.0.0 requests>=2.28.0 diff --git a/backend/routes_ai.py b/backend/routes_ai.py index 5991a3a..ae543cc 100644 --- a/backend/routes_ai.py +++ b/backend/routes_ai.py @@ -8,14 +8,44 @@ from agents.email_agent import EmailAgent from agents.meeting_agent import MeetingAgent from orchestration import chat_workflow -from app.smart_chat import SmartChatAgent from config.settings import SETTINGS import json router = APIRouter() +# SmartChatAgent (backing the /assistant/* endpoints below) was never +# committed to this repo -- `app.smart_chat` does not exist anywhere in the +# tree. That used to take the entire router down at import time, which took +# down the whole FastAPI app with it, over three endpoints out of the +# thirteen in this file. +# +# The other ten endpoints (plan_today, nudges, weekly reports, wellness +# score, email/meeting analysis, EOD reports, burnout check) don't touch +# SmartChatAgent at all and have no reason to be unavailable because of it. +# The import is optional now; only /assistant/* is affected, and it fails +# with a clear 501 rather than either fabricating a chat agent or refusing to +# boot the app. +try: + from app.smart_chat import SmartChatAgent + _SMART_CHAT_AVAILABLE = True +except ImportError: + SmartChatAgent = None # type: ignore[assignment] + _SMART_CHAT_AVAILABLE = False + # Keep a cache of SmartChatAgent instances per user -_chat_agents: Dict[str, SmartChatAgent] = {} +_chat_agents: Dict[str, Any] = {} + + +def _require_smart_chat() -> None: + if not _SMART_CHAT_AVAILABLE: + raise HTTPException( + status_code=501, + detail=( + "The conversational assistant is not available: app.smart_chat " + "was never implemented in this codebase. The other endpoints " + "in this router are unaffected." + ), + ) @router.post('/ai/plan_today') @@ -62,6 +92,7 @@ async def wellness_score(payload: dict): @router.post('/assistant/start') async def assistant_start(payload: dict): + _require_smart_chat() user_email = payload.get('user_email') if not user_email: raise HTTPException(status_code=400, detail='user_email required') @@ -75,6 +106,7 @@ async def assistant_start(payload: dict): @router.post('/assistant/chat') async def assistant_chat(payload: dict): + _require_smart_chat() session_id = payload.get('session_id') user_email = payload.get('user_email') message = payload.get('message') diff --git a/config/settings.py b/config/settings.py index a797892..62c29d0 100644 --- a/config/settings.py +++ b/config/settings.py @@ -6,21 +6,34 @@ load_dotenv() BASE_DIR = Path(__file__).resolve().parents[1] -DATA_DIR = Path(os.getenv("DATA_DIR", BASE_DIR / "data" / "mock_data_json")) + +# Opspilot's OWN generated/reference state -- tasks derived from processed +# mail, follow-up tracking, EOD/weekly reports, wellness config, governance +# audit/usage logs, the local email-processing ledger, and the known-contacts +# directory. NOT mock input data: mail and calendar content come from +# Microsoft Graph (see the "graph" block below) and are never read from here. +DATA_DIR = Path(os.getenv("DATA_DIR", BASE_DIR / "data")) SETTINGS = { "env": os.getenv("ENV", "dev"), "data": { + # Known-contacts directory (communication tone, department, etc.) -- + # Opspilot's own reference data about real people it works with, not a + # stand-in for an organizational directory. Maintain it yourself; + # starts empty. "users": DATA_DIR / "users.json", "emails": { - "inbox": DATA_DIR / "emails" / "inbox.json", - "threads": DATA_DIR / "emails" / "email_threads.json", + # Local processing-state ledger, keyed by email_id, for REAL + # emails read live from Graph. Mail content itself is never + # persisted here. + "state": DATA_DIR / "emails" / "email_state.json", }, "tasks": DATA_DIR / "tasks" / "tasks.json", "calendar": { - "meetings": DATA_DIR / "calendar" / "meetings.json", + # Previously generated MoM records -- Opspilot's own output. + # Meetings themselves come live from Graph; there is no local + # meetings.json or transcripts directory any more. "mom": DATA_DIR / "calendar" / "mom.json", - "transcripts_dir": DATA_DIR / "calendar" / "transcripts", }, "nudges": DATA_DIR / "nudges" / "followups.json", "reporting": { @@ -41,10 +54,33 @@ "policies_file": Path(os.getenv("POLICY_FILE", BASE_DIR / "governance" / "policies.json")), "daily_budget_usd": float(os.getenv("DAILY_BUDGET_USD", "50")), }, - "models": { - "chat_model": os.getenv("AZURE_OPENAI_GPT4O_MINI_DEPLOYMENT", "sc-rnd-gpt-4o-mini-01"), - "embedding_model": os.getenv("AZURE_OPENAI_EMBEDDING_DEPLOYMENT", "ai-rnd-text-embedding-ada-002"), - "azure_api_key": os.getenv("AZURE_OPENAI_API_KEY", ""), - "azure_api_base": os.getenv("AZURE_OPENAI_API_BASE", ""), + # Generic OpenAI-API-compatible LLM config. Deliberately not tied to any + # specific proxy, host, or deployment name -- point this at Azure OpenAI, + # a self-hosted LiteLLM proxy, OpenAI directly, or anything else that + # speaks the OpenAI chat-completions wire format, by setting three values. + "llm": { + "base_url": os.getenv("LLM_BASE_URL", ""), + "api_key": os.getenv("LLM_API_KEY", ""), + "model_id": os.getenv("LLM_MODEL_ID", ""), + "embedding_model_id": os.getenv("LLM_EMBEDDING_MODEL_ID", ""), + }, + # Microsoft Graph. `client_id` has no default and cannot: it must be an + # app registration YOU create (see docs/graph_setup.md) -- a shared + # default would let every deployment of this code impersonate every + # other one. `tenant="common"` allows both personal Microsoft accounts and + # work/school tenant accounts through the same app registration. + "graph": { + "client_id": os.getenv("MS_GRAPH_CLIENT_ID", ""), + "tenant": os.getenv("MS_GRAPH_TENANT", "common"), + "token_cache_path": os.getenv( + "MS_GRAPH_TOKEN_CACHE", str(BASE_DIR / ".graph_token_cache.json") + ), + }, + # LangGraph checkpointing. Redis-backed rather than in-memory: MemorySaver + # loses all workflow/conversation state on restart, silently -- the graph + # keeps running afterwards, it just has amnesia. See + # orchestration/checkpointer.py for why this matters and how it fails. + "redis": { + "url": os.getenv("REDIS_URL", "redis://localhost:6379/0"), }, } diff --git a/data/.gitkeep b/data/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/calendar/.gitkeep b/data/calendar/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/emails/.gitkeep b/data/emails/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/governance/.gitkeep b/data/governance/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/nudges/.gitkeep b/data/nudges/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/reporting/.gitkeep b/data/reporting/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/tasks/.gitkeep b/data/tasks/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/wellness/.gitkeep b/data/wellness/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/docs/graph_setup.md b/docs/graph_setup.md new file mode 100644 index 0000000..d7c474c --- /dev/null +++ b/docs/graph_setup.md @@ -0,0 +1,106 @@ +# Microsoft Graph setup + +Opspilot reads real mail and calendar data through Microsoft Graph. This +requires an app registration in the Azure Portal -- a one-time, interactive +step that only you can do, since it means signing in with your own Microsoft +account. Nothing in this codebase can create it for you. + +## Why an app registration at all + +Every application that talks to Microsoft Graph needs to identify itself. +There is no shared or default `client_id` this project could ship with -- +that would mean every deployment of this code impersonates every other one, +and Microsoft's consent screen would show your data being accessed by +"Opspilot" as if it were one single, globally shared application. Your own +registration means the consent screen shows *your* app, requesting access to +*your* data, and you control it (rotate it, revoke it, delete it) independently +of anyone else running this code. + +## Steps + +1. Go to [portal.azure.com](https://portal.azure.com) and sign in with the + Microsoft account you want Opspilot to read (personal Outlook.com account, + or a work/school account, depending on what you chose). + +2. Navigate to **Azure Active Directory** (may show as **Microsoft Entra ID**) + → **App registrations** → **New registration**. + +3. Fill in: + - **Name**: anything recognizable, e.g. `opspilot-dev`. + - **Supported account types**: choose + **"Accounts in any organizational directory and personal Microsoft + accounts"** -- this is what allows both a personal Outlook.com account + and a work/school tenant account to sign in through the same + registration. If you pick a narrower option, only one account type will + work. + - **Redirect URI**: leave blank. Device-code flow (what Opspilot uses) + doesn't redirect anywhere. + +4. After creation, note the **Application (client) ID** on the registration's + Overview page -- this is `MS_GRAPH_CLIENT_ID`. + +5. Under **Authentication**: + - Click **Add a platform** → **Mobile and desktop applications**. + - Check the box for + `https://login.microsoftonline.com/common/oauth2/nativeclient`. + - Under **Advanced settings**, set **"Allow public client flows"** to + **Yes**. This is required for device-code flow -- without it, Graph + rejects the login with an unhelpful error about the client not being + public. + +6. Under **API permissions** → **Add a permission** → **Microsoft Graph** → + **Delegated permissions**, add: + - `Mail.Read` + - `Calendars.Read` + - `OnlineMeetingTranscript.Read.All` + - `offline_access` (usually already present by default) + + Leave `User.Read` if it's there by default; Opspilot doesn't use it but it + doesn't hurt. + +## About `OnlineMeetingTranscript.Read.All` + +Add it regardless of which account type you're using -- it costs nothing to +request. What happens next depends on the account: + +- **Personal Microsoft account**: this permission is **not available** to + personal accounts at all, by Microsoft's own design. Meeting transcripts + will always come back empty, and the meeting agent falls back to generating + minutes from calendar metadata (title, agenda, attendees) instead of a real + transcript. This is a real product limitation, not a bug -- see + `repos/data_repo.py`'s `get_transcript` for where it's handled and logged. + +- **Work/school tenant account**: this is an *admin-consent* permission. A + Global Administrator (or Application Administrator) for that tenant must + grant it -- either via the same App permissions page (**Grant admin consent + for \** button, if you have the rights) or by asking whoever does. + Without that grant, transcripts will also come back empty, indistinguishable + from the personal-account case from Opspilot's point of view. + +## Completing the login + +Once `MS_GRAPH_CLIENT_ID` is set in your `.env`: + +```bash +python -m scripts.graph_login +``` + +This prints a URL and a short code. Open the URL in any browser (any device), +enter the code, sign in, and approve the requested permissions. The resulting +token is cached locally (see `MS_GRAPH_TOKEN_CACHE` in `.env`, default +`.graph_token_cache.json`) and silently refreshed on subsequent runs -- you +should not need to repeat this unless the cache is deleted or you revoke +access. + +**Do not commit `.graph_token_cache.json`.** It's already covered by +`.gitignore`; if you ever see it tracked, that's a bug, not a feature. + +## Switching accounts + +```bash +python -m scripts.graph_login --logout +``` + +Clears the local cache and removes the cached account, so the next login +starts fresh (useful for switching from a personal account to a work/school +one, or vice versa). diff --git a/governance/gateway.py b/governance/gateway.py index fe25956..f0b3787 100644 --- a/governance/gateway.py +++ b/governance/gateway.py @@ -1,23 +1,9 @@ from __future__ import annotations -import re, time, json -from typing import Dict, Any, Optional, Tuple +import re, json +from typing import Optional, Tuple from config.settings import SETTINGS -from governance.usage import write_usage - -# Import enhanced gateway -try: - from governance.litellm_gateway import EnhancedLiteLLMGateway - _enhanced_gateway_available = True -except ImportError: - _enhanced_gateway_available = False - -# OpenAI client for LiteLLM proxy (legacy fallback) -try: - from openai import OpenAI - _openai_available = True -except Exception: - _openai_available = False +from governance.litellm_gateway import EnhancedLiteLLMGateway, LLMNotConfiguredError # Load policies once POLICIES = json.loads(SETTINGS["governance"]["policies_file"].read_text()) @@ -30,338 +16,41 @@ def _redact(text: str) -> str: return text class PolicyGateway: - """ - Policy Gateway (Legacy) - - NOTE: For new code, use EnhancedLiteLLMGateway from governance.litellm_gateway - This is kept for backward compatibility. + """Thin policy wrapper over EnhancedLiteLLMGateway: agents import this, + not the gateway directly, so budget/policy checks stay in one place. + + This used to carry its own second, independent simulate/fallback + implementation -- template responses returned whenever the underlying + gateway was unconfigured or raised. That was a second copy of the same + fabricated-output problem fixed in litellm_gateway.py, and it is gone for + the same reason: a policy layer that quietly substitutes fake content for + a real answer is worse than one that fails loudly, because nothing + downstream can tell the difference. """ def __init__(self, agent_name: str): self.agent = agent_name - self.model = SETTINGS["models"]["chat_model"] - - # Try to use enhanced gateway if available - if _enhanced_gateway_available: - self._enhanced = EnhancedLiteLLMGateway(agent_name, enable_cache=True) - else: - self._enhanced = None + self._gateway = EnhancedLiteLLMGateway(agent_name, enable_cache=True) + self.model = self._gateway.model def check_budget(self) -> Tuple[bool, float]: - # NOTE: In Phase-1 we simulate budget tracking via write_usage totals (not implemented fully) - # Always allow but return remaining as default for demo + # NOTE: budget tracking is informational only right now (see + # governance/usage.py) -- this always allows and returns the + # configured ceiling, it does not enforce a running total. agent_budget = POLICIES.get("daily_budgets", {}).get(f"{self.agent}_usd", DAILY_BUDGET_DEFAULT) return True, agent_budget def call_llm(self, prompt: str, temperature: float = 0.2, - max_tokens: int = 1024, correlation_id: Optional[str] = None) -> str: - """ - Call LLM with policy enforcement - - This method now delegates to EnhancedLiteLLMGateway if available, - which provides caching, retry logic, and better error handling. - """ - import time - start = time.time() - - # Use enhanced gateway if available - if self._enhanced: - try: - return self._enhanced.call( - prompt=prompt, - temperature=temperature, - max_tokens=max_tokens, - correlation_id=correlation_id, - use_cache=True - ) - except Exception as e: - # Fallback to legacy implementation - print(f"⚠️ Enhanced gateway failed, falling back to legacy: {e}") - - ok, remaining = self.check_budget() - if not ok: - write_usage(self.agent, self.model, 0, 0, 0, 0.0, "fail", rate_limited=True, correlation_id=correlation_id) - raise RuntimeError(f"Budget exceeded for {self.agent}") - - # IMPORTANT: Redact only for LOGGING, not for the prompt used in deterministic fallback - safe_prompt_for_logging = _redact(prompt) - - # ---- Simulation mode (no API key or openai not installed) ---- - if not _openai_available or not SETTINGS["models"]["azure_api_key"]: - # Generate a deterministic but proper output depending on agent - out = self._simulate_response(prompt) - # Log usage (simulated) - write_usage(self.agent, self.model, len(prompt)//4, len(out)//4, - int((time.time()-start)*1000), 0.0, "success", correlation_id=correlation_id, - meta={"simulated": True}) - return out - - # ---- Real LLM call via OpenAI client to LiteLLM proxy ---- - # The proxy expects the full model name (e.g., azure/sc-rnd-gpt-4o-mini-01) - try: - client = OpenAI( - base_url=SETTINGS["models"]["azure_api_base"], - api_key=SETTINGS["models"]["azure_api_key"] - ) - resp = client.chat.completions.create( - model=self.model, # e.g., "azure/sc-rnd-gpt-4o-mini-01" - messages=[ - {"role": "system", "content": f"You are the {self.agent} with enterprise guardrails."}, - {"role": "user", "content": prompt} - ], - temperature=temperature, - max_tokens=max_tokens, - timeout=30.0 - ) - text = resp.choices[0].message.content - usage = resp.usage - tokens_in = usage.prompt_tokens if usage else 0 - tokens_out = usage.completion_tokens if usage else 0 - latency = int((time.time() - start) * 1000) - write_usage(self.agent, self.model, tokens_in, tokens_out, latency, 0.0, "success", - correlation_id=correlation_id, - meta={"prompt_redacted_for_logs": bool(safe_prompt_for_logging != prompt)}) - return text - except Exception as e: - write_usage(self.agent, self.model, 0, 0, int((time.time()-start)*1000), 0.0, "fail", - correlation_id=correlation_id, meta={"error": str(e)}) - raise - - # Add this helper method inside PolicyGateway class: - def _simulate_response(self, prompt: str) -> str: - """ - Deterministic, template-based responses for Phase-1 demo when no Azure key is configured. - Produces clean outputs instead of echoing the prompt. - """ - # Email agent templates - if self.agent == "email_agent": - # Heuristics: if the prompt includes 'Write a concise, professional reply' -> return a real reply - if "Write a concise, professional reply" in prompt: - return ( - "Hi team,\n\n" - "Thanks for the update on the quarterly noise reduction initiative. I acknowledge the plan for regular checks " - "and the awareness campaign. I’ll participate and ensure my area follows the guidelines.\n\n" - "Next steps:\n" - "• Note the schedule for noise checks\n" - "• Share the awareness materials with the team\n" - "• Raise any issues to the supervisor as needed\n\n" - "Best regards,\n" - "Kowshik Naidu\n" - "SmartOps Engineer" - ) - if "Summarize the email" in prompt: - return ("Summary: The manager announces a quarterly noise reduction initiative with regular checks and an " - "awareness campaign; team cooperation is requested and queries should go to supervisors.") - if "From the email, extract explicit action items" in prompt: - # Nothing actionable in this ‘noise’ example - return "[]" - - # Meeting agent templates - if self.agent == "meeting_agent": - # Parse transcript to generate dynamic MoM - return self._generate_dynamic_mom(prompt) - - # Tasks/Followup/Reporting simple defaults - if self.agent == "tasks_agent": - return "Focus on P0/P1 items first; allocate 2–3 focused blocks. Communicate blockers early." - if self.agent == "followup_agent": - import re - - def extract(field): - m = re.search(fr"{field}:(.*)", prompt) - return (m.group(1).strip() if m else "") - - title = extract("Task Title") - due_date = extract("Due Date") - status = extract("Status") - priority = extract("Priority") - - return ( - f"Quick reminder about {title}. " - f"It is currently {status}, with a priority of {priority}, " - f"and the due date is {due_date}. " - f"Could you share a short update or let me know if anything is blocking progress?" - ) - if self.agent == "reporting_agent": - return "Completed: 1; In progress: 1; Pending: 1. Risks noted; follow-ups initiated." - - # Chat router - intent classification - if self.agent == "chat_router": - # Parse the intent classification request - import json - prompt_lower = prompt.lower() - - # Default intent detection based on keywords - if any(kw in prompt_lower for kw in ['p0', 'p1', 'p2', 'task', 'todo', 'overdue', 'my tasks']): - return json.dumps({ - "intent": "tasks.list", - "confidence": 0.95, - "slots": {"priority": "P0" if "p0" in prompt_lower else None}, - "reasoning": "Detected task-related keywords" - }) - elif any(kw in prompt_lower for kw in ['email', 'inbox', 'unread', 'mail']): - return json.dumps({ - "intent": "emails.list", - "confidence": 0.9, - "slots": {}, - "reasoning": "Detected email-related keywords" - }) - elif any(kw in prompt_lower for kw in ['meeting', 'calendar', 'schedule', 'agenda']): - return json.dumps({ - "intent": "meetings.list", - "confidence": 0.9, - "slots": {}, - "reasoning": "Detected meeting-related keywords" - }) - elif any(kw in prompt_lower for kw in ['brief', 'summary', "what's up", 'catch up']): - return json.dumps({ - "intent": "briefing", - "confidence": 0.95, - "slots": {}, - "reasoning": "User wants a briefing" - }) - elif any(kw in prompt_lower for kw in ['followup', 'follow-up', 'follow up', 'pending']): - return json.dumps({ - "intent": "followups.list", - "confidence": 0.9, - "slots": {}, - "reasoning": "Detected follow-up keywords" - }) - elif any(kw in prompt_lower for kw in ['wellness', 'stress', 'tired', 'burnout', 'break']): - return json.dumps({ - "intent": "wellness.score", - "confidence": 0.85, - "slots": {}, - "reasoning": "Detected wellness-related keywords" - }) - elif any(kw in prompt_lower for kw in ['hello', 'hi ', 'hey', 'good morning', 'good afternoon']): - return json.dumps({ - "intent": "greeting", - "confidence": 1.0, - "slots": {}, - "reasoning": "User greeted" - }) - else: - return json.dumps({ - "intent": "general", - "confidence": 0.7, - "slots": {}, - "reasoning": "General query" - }) - - # Smart chat agent responses - if self.agent == "smart_chat": - return "I'll help you with that request." - - # Generic fallback - return "Acknowledged." - - def _generate_dynamic_mom(self, prompt: str) -> str: - """ - Parse transcript from prompt and generate detailed, dynamic MoM. - Extracts actual content from the meeting transcript. - """ - import re - import json - - # Extract transcript from prompt - transcript_match = re.search(r'Transcript:\s*(.+)', prompt, re.DOTALL) - transcript = transcript_match.group(1).strip() if transcript_match else "" - - if not transcript: - return json.dumps({ - "summary": "No transcript available for this meeting.", - "decisions": [], - "action_items": [], - "risks": [], - "dependencies": [] - }) - - lines = transcript.split('\n') - - # Extract participants from "Speaker X (Name, Company):" patterns - participants = set() - for line in lines: - match = re.match(r'Speaker \d+ \(([^)]+)\):', line) - if match: - participants.add(match.group(1).split(',')[0].strip()) - - # Identify key topics by looking for important keywords - topics = [] - decisions = [] - action_items = [] - risks = [] - dependencies = [] - - transcript_lower = transcript.lower() - - # Extract specific content patterns - for line in lines: - line_lower = line.lower() - # Clean up speaker prefix for all extractions - clean = re.sub(r'^Speaker \d+ \([^)]+\):\s*', '', line).strip() - clean = re.sub(r'^Speaker \d+:\s*', '', clean).strip() - - # Look for decisions (keywords: decided, agreed, approved, confirmed, let's do) - if any(kw in line_lower for kw in ['decided', 'agreed', 'approved', 'confirmed', "let's do", 'we\'ll go with', 'that\'s the plan']): - if clean and len(clean) > 20: - decisions.append(clean[:200]) - - # Look for action items (keywords: I'll, we'll, will send, by Friday, by end of, need to) - if any(kw in line_lower for kw in ["i'll", "we'll", "will send", "by friday", "by end of", "i will", "we will", "let me", "i can have"]): - if clean and len(clean) > 15: - action_items.append(clean[:200]) - - # Look for risks (keywords: risk, concern, worried, might, could fail, blocker, delay) - if any(kw in line_lower for kw in ['risk', 'concern', 'worried', 'might be', 'could fail', 'blocker', 'delay', 'tight', 'challenge']): - if clean and len(clean) > 15: - risks.append(clean[:200]) - - # Look for dependencies (keywords: need access, waiting for, depends on, requires) - if any(kw in line_lower for kw in ['need access', 'waiting for', 'depends on', 'requires', 'need from', 'connect you']): - if clean and len(clean) > 15: - dependencies.append(clean[:200]) - - # Extract key metrics/numbers mentioned - metrics = re.findall(r'\$[\d,]+k?|\d+(?:\.\d+)?%|\d+-\d+ (?:months?|weeks?|days?)|\d+ (?:months?|weeks?|days?)', transcript) - - # Build summary from first few substantive lines and participants - summary_parts = [] - if participants: - summary_parts.append(f"Meeting between {', '.join(sorted(list(participants))[:4])}.") - - # Find key discussion points from early lines (skip greetings) - skip_starts = ('thanks', 'thank you', 'welcome', 'hi ', 'hello', 'good morning', 'good afternoon') - for line in lines[:20]: - clean = re.sub(r'^Speaker \d+ \([^)]+\):\s*', '', line).strip() - clean = re.sub(r'^Speaker \d+:\s*', '', clean).strip() - if len(clean) > 40 and not clean.lower().startswith(skip_starts): - summary_parts.append(clean[:180]) - if len(summary_parts) >= 3: - break - break - - if metrics: - summary_parts.append(f"Key figures discussed: {', '.join(metrics[:5])}.") - - summary = ' '.join(summary_parts) if summary_parts else "Meeting discussion captured." - - # Deduplicate and limit - def dedupe(items, limit=5): - seen = set() - result = [] - for item in items: - key = item.lower()[:50] - if key not in seen: - seen.add(key) - result.append(item) - if len(result) >= limit: - break - return result - - return json.dumps({ - "summary": summary[:500], - "decisions": dedupe(decisions), - "action_items": dedupe(action_items), - "risks": dedupe(risks), - "dependencies": dedupe(dependencies) - }) \ No newline at end of file + max_tokens: int = 1024, correlation_id: Optional[str] = None) -> str: + """Call the LLM with policy enforcement. Raises on any failure -- + misconfiguration (LLMNotConfiguredError) or a real API error -- rather + than returning a fabricated response.""" + # Redaction is for the audit log only; the prompt sent to the model + # is never altered by it. + _redact(prompt) + + return self._gateway.call( + prompt=prompt, + temperature=temperature, + max_tokens=max_tokens, + correlation_id=correlation_id, + ) diff --git a/governance/litellm_gateway.py b/governance/litellm_gateway.py index c464243..1bc6985 100644 --- a/governance/litellm_gateway.py +++ b/governance/litellm_gateway.py @@ -1,22 +1,33 @@ # governance/litellm_gateway.py """ -Enhanced LiteLLM Gateway -======================== -Production-grade LLM gateway with: -- Response caching (reduces costs by 90% on repeated queries) -- Automatic retry with exponential backoff -- Streaming support for real-time responses -- Token budget enforcement -- Request/response logging -- Error handling and fallbacks - -Optimized for single model (gpt-4o-mini) with smart prompt engineering -to maximize quality despite model constraints. +LLM Gateway +=========== +Production-grade LLM gateway: response caching, retry with backoff, streaming, +structured-output parsing, usage/cost logging. + +Name kept for backward compatibility -- seven orchestration modules import +`EnhancedLiteLLMGateway` from this path. It has never actually used the +`litellm` package (see requirements.txt); it always talked to an OpenAI-API- +compatible endpoint directly via the `openai` SDK. + +Config is now fully generic: `LLM_BASE_URL` / `LLM_API_KEY` / `LLM_MODEL_ID` +point this at anything speaking the OpenAI chat-completions wire format -- +Azure OpenAI, a self-hosted LiteLLM proxy, OpenAI directly, whatever. Nothing +here is hardcoded to one deployment name or one proxy host. + +Fails closed. This used to have THREE separate layers of fabricated fallback +responses -- PolicyGateway's own simulate, this class's simulate, and +PolicyGateway re-simulating if this class raised -- all returning literal +placeholder content like `"decisions": ["Decision 1", "Decision 2"]` when no +API key was configured, indistinguishable from a real analysis to whatever +called it. All three are gone. Missing config raises LLMNotConfiguredError at +construction time; a real API failure after retries raises, it does not return +a canned string that reads like an answer. """ from __future__ import annotations -from typing import Optional, Dict, Any, List, Generator, Union -from datetime import datetime, timedelta +from typing import Optional, Dict, Any, Generator, Union +from datetime import datetime, timedelta, timezone import json import time import hashlib @@ -27,13 +38,21 @@ from governance.usage import write_usage +class LLMNotConfiguredError(RuntimeError): + """LLM_BASE_URL / LLM_API_KEY / LLM_MODEL_ID are not set. + + Raised at gateway construction, not on first call -- a misconfigured + gateway should fail when it's built, not three steps into a workflow when + an agent finally tries to use it. + """ + + # ============================================================ -# CACHING SYSTEM +# CACHING # ============================================================ @dataclass class CacheEntry: - """Cached LLM response""" prompt_hash: str response: str timestamp: str @@ -44,61 +63,46 @@ class CacheEntry: class DiskCache: - """Simple disk-based cache for LLM responses""" - - def __init__(self, cache_dir: str = "./.llm_cache"): + """Disk-based cache for LLM responses, keyed by (prompt, model, temperature).""" + + def __init__(self, cache_dir: str = "./.llm_cache", ttl_hours: int = 24): self.cache_dir = Path(cache_dir) self.cache_dir.mkdir(exist_ok=True) - self.ttl_hours = 24 # Cache expires after 24 hours - + self.ttl_hours = ttl_hours + def _hash_prompt(self, prompt: str, model: str, temperature: float) -> str: - """Create hash of prompt + params for cache key""" key = f"{prompt}|{model}|{temperature}" return hashlib.sha256(key.encode()).hexdigest()[:16] - + def get(self, prompt: str, model: str, temperature: float) -> Optional[str]: - """Retrieve cached response if exists and not expired""" - cache_key = self._hash_prompt(prompt, model, temperature) - cache_file = self.cache_dir / f"{cache_key}.json" - + cache_file = self.cache_dir / f"{self._hash_prompt(prompt, model, temperature)}.json" if not cache_file.exists(): return None - try: data = json.loads(cache_file.read_text(encoding="utf-8")) - - # Check expiry cached_time = datetime.fromisoformat(data["timestamp"]) - age = datetime.utcnow() - cached_time - - if age > timedelta(hours=self.ttl_hours): - cache_file.unlink() # Delete expired cache + if datetime.now(timezone.utc) - cached_time > timedelta(hours=self.ttl_hours): + cache_file.unlink() return None - return data["response"] except Exception: return None - - def set(self, prompt: str, model: str, temperature: float, - response: str, tokens_in: int, tokens_out: int, metadata: Dict = None): - """Store response in cache""" - cache_key = self._hash_prompt(prompt, model, temperature) - cache_file = self.cache_dir / f"{cache_key}.json" - + + def set(self, prompt: str, model: str, temperature: float, response: str, + tokens_in: int, tokens_out: int, metadata: Optional[Dict] = None): + cache_file = self.cache_dir / f"{self._hash_prompt(prompt, model, temperature)}.json" entry = CacheEntry( - prompt_hash=cache_key, + prompt_hash=cache_file.stem, response=response, - timestamp=datetime.utcnow().isoformat(), + timestamp=datetime.now(timezone.utc).isoformat(), model=model, tokens_in=tokens_in, tokens_out=tokens_out, - metadata=metadata or {} + metadata=metadata or {}, ) - cache_file.write_text(json.dumps(entry.__dict__, indent=2), encoding="utf-8") - + def clear(self): - """Clear all cached entries""" for cache_file in self.cache_dir.glob("*.json"): cache_file.unlink() @@ -108,116 +112,100 @@ def clear(self): # ============================================================ class PromptOptimizer: - """ - Optimize prompts for single model (gpt-4o-mini) - Adds techniques to improve quality without multiple models: - - Chain-of-thought prompting - - Role-specific system messages - - Output format constraints - """ - + """Role-specific system messages and output-format constraints.""" + ROLE_SYSTEM_MESSAGES = { "email_agent": """You are an expert email analyst and communication specialist. Your role: Analyze emails, extract action items, draft professional responses. Approach: Be concise, actionable, and context-aware. Output: Always provide structured, JSON-compatible responses when requested.""", - + "meeting_agent": """You are a meeting intelligence specialist. Your role: Analyze meeting transcripts, extract decisions, identify action items. Approach: Be specific - use actual names, dates, and technical details from transcripts. Output: Provide detailed, structured summaries with clear ownership.""", - + "tasks_agent": """You are a productivity and task management expert. Your role: Analyze workload, prioritize tasks, create focus plans. Approach: Consider urgency, dependencies, and user capacity. Output: Provide actionable daily plans with time blocks.""", - + "wellness_agent": """You are a workplace wellness and burnout prevention specialist. Your role: Monitor workload, detect stress signals, suggest interventions. Approach: Be empathetic yet data-driven, focus on sustainable productivity. Output: Provide wellness scores with actionable recommendations.""", - + "chat_agent": """You are an intelligent workplace assistant. Your role: Help users with tasks, emails, meetings, and wellness through conversation. Approach: Be conversational but efficient, ask clarifying questions when needed. Output: Provide helpful, contextual responses based on user's data.""", } - + @classmethod - def enhance_prompt(cls, agent_name: str, prompt: str, - output_format: Optional[str] = None) -> tuple[str, str]: - """ - Enhance prompt with CoT and format instructions - Returns: (system_message, enhanced_prompt) - """ + def enhance_prompt(cls, agent_name: str, prompt: str, + output_format: Optional[str] = None) -> tuple[str, str]: system_msg = cls.ROLE_SYSTEM_MESSAGES.get( agent_name, f"You are {agent_name}, a helpful AI assistant for workplace automation." ) - + enhanced = prompt - - # Add Chain-of-Thought for complex reasoning - if any(keyword in prompt.lower() for keyword in ["analyze", "extract", "decide", "plan"]): - enhanced = f"""Think step by step: -1. First, understand the context and requirements -2. Then, identify key information -3. Finally, provide your response - -Task: -{prompt}""" - - # Add output format constraints + if any(kw in prompt.lower() for kw in ["analyze", "extract", "decide", "plan"]): + enhanced = ( + "Think step by step:\n" + "1. First, understand the context and requirements\n" + "2. Then, identify key information\n" + "3. Finally, provide your response\n\n" + f"Task:\n{prompt}" + ) + if output_format: enhanced += f"\n\nIMPORTANT: Return ONLY valid {output_format}. No markdown, no code fences, no extra text." - + return system_msg, enhanced # ============================================================ -# ENHANCED LITELLM GATEWAY +# GATEWAY # ============================================================ class EnhancedLiteLLMGateway: + """LLM gateway: caching, retry, streaming, structured output, usage logging. + + Talks to whatever OpenAI-API-compatible endpoint `SETTINGS["llm"]` + describes. There is no simulation mode -- construction raises + LLMNotConfiguredError if base_url/api_key/model_id are missing, and `call` + raises after the retry budget is exhausted rather than returning a + template string. """ - Production-grade LiteLLM gateway optimized for single model deployment - - Features: - - Smart caching to reduce costs - - Retry logic with exponential backoff - - Streaming for real-time UX - - Budget enforcement - - Comprehensive error handling - """ - + def __init__( self, agent_name: str, - daily_budget_usd: float = None, + daily_budget_usd: Optional[float] = None, enable_cache: bool = True, - enable_retry: bool = True + max_retries: int = 3, ): self.agent_name = agent_name self.daily_budget = daily_budget_usd or SETTINGS["governance"]["daily_budget_usd"] - self.model = SETTINGS["models"]["chat_model"] - self.base_url = SETTINGS["models"]["azure_api_base"] - self.api_key = SETTINGS["models"]["azure_api_key"] - + + llm_cfg = SETTINGS["llm"] + self.model = llm_cfg["model_id"] + self.base_url = llm_cfg["base_url"] + self.api_key = llm_cfg["api_key"] + + if not (self.base_url and self.api_key and self.model): + raise LLMNotConfiguredError( + "LLM_BASE_URL, LLM_API_KEY, and LLM_MODEL_ID must all be set. " + "Point them at any OpenAI-API-compatible endpoint -- Azure OpenAI, " + "a self-hosted LiteLLM proxy, OpenAI directly, etc." + ) + + self.client = OpenAI(base_url=self.base_url, api_key=self.api_key) self.cache = DiskCache() if enable_cache else None - self.enable_retry = enable_retry - self.max_retries = 3 - + self.max_retries = max_retries self.optimizer = PromptOptimizer() - - # Initialize OpenAI client for LiteLLM proxy - if self.api_key: - self.client = OpenAI( - base_url=self.base_url, - api_key=self.api_key - ) - else: - self.client = None - + def call( self, prompt: str, @@ -225,265 +213,134 @@ def call( max_tokens: int = 1024, stream: bool = False, use_cache: bool = True, - output_format: Optional[str] = None, # "JSON" | "text" + output_format: Optional[str] = None, correlation_id: Optional[str] = None, - **kwargs + **kwargs, ) -> Union[str, Generator[str, None, None]]: - """ - Make LLM call with optimizations - - Args: - prompt: The prompt to send - temperature: Sampling temperature (0.0 = deterministic) - max_tokens: Maximum tokens in response - stream: Whether to stream response - use_cache: Whether to use cached responses - output_format: Expected output format for validation - correlation_id: For tracking related operations - - Returns: - String response or generator if streaming - """ start_time = time.time() - - # Check cache first (only for non-streaming) + if use_cache and not stream and self.cache: cached = self.cache.get(prompt, self.model, temperature) if cached: - # Log cache hit - write_usage( - self.agent_name, - self.model, - 0, 0, 0, 0.0, - "success", - correlation_id=correlation_id, - meta={"cache_hit": True} - ) + write_usage(self.agent_name, self.model, 0, 0, 0, 0.0, "success", + correlation_id=correlation_id, meta={"cache_hit": True}) return cached - - # Check budget - if not self._check_budget(): - raise RuntimeError(f"Daily budget exceeded for {self.agent_name}") - - # Optimize prompt + system_msg, enhanced_prompt = self.optimizer.enhance_prompt( - self.agent_name, - prompt, - output_format + self.agent_name, prompt, output_format ) - - # Simulation mode (no API key) - if not self.client: - return self._simulate_response(enhanced_prompt, output_format) - - # Make real API call with retry + + last_error: Optional[Exception] = None for attempt in range(self.max_retries): try: response = self.client.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": system_msg}, - {"role": "user", "content": enhanced_prompt} + {"role": "user", "content": enhanced_prompt}, ], temperature=temperature, max_tokens=max_tokens, stream=stream, - timeout=30.0 + timeout=30.0, ) - + if stream: return self._handle_streaming(response, correlation_id) - else: - content = response.choices[0].message.content - - # Log usage - usage = response.usage - latency_ms = int((time.time() - start_time) * 1000) - self._log_usage( - usage.prompt_tokens, - usage.completion_tokens, - latency_ms, - correlation_id + + content = response.choices[0].message.content + usage = response.usage + latency_ms = int((time.time() - start_time) * 1000) + self._log_usage( + usage.prompt_tokens if usage else 0, + usage.completion_tokens if usage else 0, + latency_ms, + correlation_id, + ) + + if use_cache and self.cache: + self.cache.set( + prompt, self.model, temperature, content, + usage.prompt_tokens if usage else 0, + usage.completion_tokens if usage else 0, + {"correlation_id": correlation_id}, ) - - # Cache response - if use_cache and self.cache: - self.cache.set( - prompt, self.model, temperature, - content, - usage.prompt_tokens, - usage.completion_tokens, - {"correlation_id": correlation_id} - ) - - return content - + return content + except Exception as e: + last_error = e if attempt == self.max_retries - 1: - # Last attempt failed self._log_error(str(e), correlation_id) raise - - # Exponential backoff - wait_time = 2 ** attempt - time.sleep(wait_time) - - raise RuntimeError("All retry attempts failed") - + time.sleep(2 ** attempt) # exponential backoff + + # Unreachable in practice -- the loop above always returns or raises -- + # but keeps the function's control flow explicit rather than implying + # a None return is possible. + raise RuntimeError(f"All retry attempts failed: {last_error}") + def call_structured( self, prompt: str, schema: Dict[str, Any], temperature: float = 0.0, - correlation_id: Optional[str] = None + correlation_id: Optional[str] = None, ) -> Dict[str, Any]: - """ - Call LLM and parse JSON response - - Args: - prompt: The prompt - schema: Expected JSON schema (for documentation) - temperature: Use 0.0 for deterministic structured output - - Returns: - Parsed JSON object - """ schema_str = json.dumps(schema, indent=2) - enhanced_prompt = f"""{prompt} - -Return your response as valid JSON matching this schema: -{schema_str} + enhanced_prompt = ( + f"{prompt}\n\nReturn your response as valid JSON matching this schema:\n" + f"{schema_str}\n\n" + "CRITICAL: Return ONLY the JSON object. No markdown, no code fences, no explanations." + ) -CRITICAL: Return ONLY the JSON object. No markdown, no code fences, no explanations.""" - response = self.call( prompt=enhanced_prompt, temperature=temperature, output_format="JSON", - correlation_id=correlation_id + correlation_id=correlation_id, ) - - # Parse JSON with error handling + + clean_response = response.strip() + if clean_response.startswith("```"): + lines = clean_response.split("\n") + clean_response = "\n".join(lines[1:-1]) if len(lines) > 2 else clean_response + try: - # Clean up response (remove markdown if present) - clean_response = response.strip() - if clean_response.startswith("```"): - # Extract JSON from markdown code block - lines = clean_response.split("\n") - clean_response = "\n".join(lines[1:-1]) if len(lines) > 2 else clean_response - return json.loads(clean_response) except json.JSONDecodeError as e: - # Fallback: Try to extract JSON from text import re - json_match = re.search(r'\{.*\}', response, re.DOTALL) + json_match = re.search(r"\{.*\}", response, re.DOTALL) if json_match: try: return json.loads(json_match.group(0)) - except: + except Exception: pass - raise ValueError(f"Failed to parse JSON response: {e}\nResponse: {response[:200]}") - - def _handle_streaming( - self, - response, - correlation_id: Optional[str] - ) -> Generator[str, None, None]: - """Handle streaming response""" + + def _handle_streaming(self, response, correlation_id: Optional[str]) -> Generator[str, None, None]: full_content = [] - for chunk in response: if chunk.choices[0].delta.content: content = chunk.choices[0].delta.content full_content.append(content) yield content - - # Log usage after streaming completes - self._log_usage( - 0, len("".join(full_content)) // 4, # Rough token estimate - 0, - correlation_id, - meta={"streaming": True} - ) - - def _check_budget(self) -> bool: - """Check if within daily budget""" - # For Phase 1, always allow (budget tracking to be enhanced) - return True - - def _log_usage( - self, - tokens_in: int, - tokens_out: int, - latency_ms: int, - correlation_id: Optional[str], - meta: Dict = None - ): - """Log token usage""" - # Approximate cost (adjust based on actual pricing) - cost_per_1k = 0.00015 # $0.15 per 1M tokens for GPT-4o-mini + self._log_usage(0, len("".join(full_content)) // 4, 0, correlation_id, meta={"streaming": True}) + + def _log_usage(self, tokens_in: int, tokens_out: int, latency_ms: int, + correlation_id: Optional[str], meta: Optional[Dict] = None): + cost_per_1k = 0.00015 # rough estimate; adjust for the configured model's real pricing cost = (tokens_in + tokens_out) / 1000 * cost_per_1k - - write_usage( - self.agent_name, - self.model, - tokens_in, - tokens_out, - latency_ms, - cost, - "success", - correlation_id=correlation_id, - meta=meta or {} - ) - - def _log_error(self, error: str, correlation_id: Optional[str]): - """Log error""" - write_usage( - self.agent_name, - self.model, - 0, 0, 0, 0.0, - "fail", - correlation_id=correlation_id, - meta={"error": error} - ) - - def _simulate_response(self, prompt: str, output_format: Optional[str]) -> str: - """Simulate response when no API key (for demo)""" - # Return reasonable defaults based on agent - if output_format == "JSON": - if "email" in prompt.lower(): - return json.dumps({ - "category": "actionable", - "priority": "P1", - "actions": ["Review and respond"], - "summary": "This email requires your attention." - }) - elif "meeting" in prompt.lower(): - return json.dumps({ - "summary": "Meeting discussion summary", - "decisions": ["Decision 1", "Decision 2"], - "action_items": ["Action 1", "Action 2"], - "risks": [], - "dependencies": [] - }) - else: - return json.dumps({"result": "Simulated response"}) - else: - return f"This is a simulated response for demo purposes. The {self.agent_name} would normally provide detailed analysis here." + write_usage(self.agent_name, self.model, tokens_in, tokens_out, latency_ms, + cost, "success", correlation_id=correlation_id, meta=meta or {}) + def _log_error(self, error: str, correlation_id: Optional[str]): + write_usage(self.agent_name, self.model, 0, 0, 0, 0.0, "fail", + correlation_id=correlation_id, meta={"error": error}) -# ============================================================ -# CONVENIENCE FUNCTIONS -# ============================================================ def create_gateway(agent_name: str, **kwargs) -> EnhancedLiteLLMGateway: - """Factory function to create gateway""" return EnhancedLiteLLMGateway(agent_name, **kwargs) def clear_cache(): - """Clear all cached LLM responses""" - cache = DiskCache() - cache.clear() - print("✅ LLM cache cleared") + DiskCache().clear() diff --git a/orchestration/autonomous_graph.py b/orchestration/autonomous_graph.py index 20eed06..1f487d7 100644 --- a/orchestration/autonomous_graph.py +++ b/orchestration/autonomous_graph.py @@ -18,7 +18,7 @@ import operator from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver +from orchestration.checkpointer import get_checkpointer # Phase 1 imports - Memory & Enhanced Gateway from memory import AgentMemory, MemoryType, EpisodicMemory, EpisodeType, EpisodeOutcome @@ -736,7 +736,7 @@ def build_email_processing_graph(): ) # Compile with memory for checkpointing - memory = MemorySaver() + memory = get_checkpointer() return graph.compile(checkpointer=memory) diff --git a/orchestration/chat_workflow.py b/orchestration/chat_workflow.py index fb7f288..b7b678a 100644 --- a/orchestration/chat_workflow.py +++ b/orchestration/chat_workflow.py @@ -20,7 +20,6 @@ import uuid from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver from governance.litellm_gateway import EnhancedLiteLLMGateway from memory import AgentMemory, EpisodicMemory, MemoryType, EpisodeType, EpisodeOutcome diff --git a/orchestration/checkpointer.py b/orchestration/checkpointer.py new file mode 100644 index 0000000..2160f3d --- /dev/null +++ b/orchestration/checkpointer.py @@ -0,0 +1,78 @@ +"""Shared checkpointer factory for every orchestration graph. + +Redis-backed rather than the in-process `MemorySaver` every graph used +before. `MemorySaver` keeps checkpoint state in a plain Python dict inside +the running process -- a restart, redeploy, or crash discards every +in-flight conversation and workflow state with no warning. The graph keeps +running afterwards; it just has amnesia. That is the same class of problem +as an in-memory rate limiter or an in-memory privacy budget: it looks like +persistence until the moment the process restarts, and demos never restart +mid-conversation so the gap goes unnoticed until production does. + +One shared saver, not one per graph: `RedisSaver` holds a live connection, and +five graphs each opening their own would mean five connections and five +independent `.setup()` calls doing the same idempotent index creation. All +five graphs safely share one. + +Requires a real, reachable Redis. There's no way to fake persistence in-process +the way MemorySaver did, so a missing/unreachable Redis is deliberately a +loud, immediate error at construction time -- not a silent fall-through to +in-memory storage, which would quietly reintroduce the exact problem this +module exists to fix. +""" + +from __future__ import annotations + +import logging +import threading + +from config.settings import SETTINGS + +logger = logging.getLogger(__name__) + +_saver = None +_setup_done = False +_lock = threading.Lock() + + +class CheckpointerUnavailable(RuntimeError): + """Redis could not be reached. Run one locally, e.g. `docker run -p 6379:6379 redis`, + or point REDIS_URL at an existing instance.""" + + +def get_checkpointer(): + """Return the shared Redis-backed checkpointer, constructing it on first use. + + Raises CheckpointerUnavailable with a clear, actionable message if Redis + is not reachable -- the underlying redis-py ConnectionError is accurate + but unhelpful on its own ("Error 10061 connecting to localhost:6379" + doesn't tell you what to do about it). + """ + global _saver, _setup_done + if _saver is not None: + return _saver + + with _lock: + if _saver is not None: + return _saver + + from langgraph.checkpoint.redis import RedisSaver + + redis_url = SETTINGS["redis"]["url"] + try: + saver = RedisSaver(redis_url=redis_url) + except Exception as e: + raise CheckpointerUnavailable( + f"Could not connect to Redis at {redis_url}: {e}. " + "Run one locally (`docker run -p 6379:6379 redis`) or set " + "REDIS_URL to a reachable instance." + ) from e + + if not _setup_done: + # Idempotent: creates the RediSearch indices checkpointing needs + # if they don't already exist. Safe to call every process start. + saver.setup() + _setup_done = True + + _saver = saver + return _saver diff --git a/orchestration/followup_reporting_subgraphs.py b/orchestration/followup_reporting_subgraphs.py index 76b8b36..4217f23 100644 --- a/orchestration/followup_reporting_subgraphs.py +++ b/orchestration/followup_reporting_subgraphs.py @@ -16,7 +16,6 @@ import operator from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver # Phase 1 imports from memory import AgentMemory, MemoryType, EpisodicMemory, EpisodeOutcome diff --git a/orchestration/meeting_subgraph.py b/orchestration/meeting_subgraph.py index f391e15..96e9e1f 100644 --- a/orchestration/meeting_subgraph.py +++ b/orchestration/meeting_subgraph.py @@ -21,7 +21,7 @@ import re from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver +from orchestration.checkpointer import get_checkpointer # Phase 1 imports from memory import AgentMemory, MemoryType, EpisodicMemory, EpisodeType, EpisodeOutcome @@ -557,7 +557,7 @@ def create_meeting_workflow() -> StateGraph: def create_meeting_workflow_with_memory() -> StateGraph: """Create meeting workflow with memory persistence""" graph = create_meeting_workflow() - memory = MemorySaver() + memory = get_checkpointer() return graph.compile(checkpointer=memory) diff --git a/orchestration/super_graph.py b/orchestration/super_graph.py index ebeab09..8b8b5c7 100644 --- a/orchestration/super_graph.py +++ b/orchestration/super_graph.py @@ -19,7 +19,7 @@ import json from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver +from orchestration.checkpointer import get_checkpointer # Phase 1 imports from memory import AgentMemory, MemoryType, EpisodicMemory, EpisodeType, EpisodeOutcome @@ -765,7 +765,7 @@ def create_super_graph() -> StateGraph: def create_super_graph_with_memory() -> StateGraph: """Create super-graph with memory persistence""" graph = create_super_graph() - memory = MemorySaver() + memory = get_checkpointer() return graph.compile(checkpointer=memory) diff --git a/orchestration/task_subgraph.py b/orchestration/task_subgraph.py index 769b0ce..91e0f75 100644 --- a/orchestration/task_subgraph.py +++ b/orchestration/task_subgraph.py @@ -20,7 +20,7 @@ import json from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver +from orchestration.checkpointer import get_checkpointer # Phase 1 imports from memory import AgentMemory, MemoryType, EpisodicMemory, EpisodeType, EpisodeOutcome @@ -536,7 +536,7 @@ def create_task_workflow() -> StateGraph: def create_task_workflow_with_memory() -> StateGraph: """Create task workflow with memory persistence""" graph = create_task_workflow() - memory = MemorySaver() + memory = get_checkpointer() return graph.compile(checkpointer=memory) diff --git a/orchestration/wellness_subgraph.py b/orchestration/wellness_subgraph.py index 4f66064..ea5c713 100644 --- a/orchestration/wellness_subgraph.py +++ b/orchestration/wellness_subgraph.py @@ -21,7 +21,7 @@ import json from langgraph.graph import StateGraph, END -from langgraph.checkpoint.memory import MemorySaver +from orchestration.checkpointer import get_checkpointer # Phase 1 imports from memory import AgentMemory, MemoryType, EpisodicMemory, EpisodeType, EpisodeOutcome @@ -556,7 +556,7 @@ def create_wellness_workflow() -> StateGraph: def create_wellness_workflow_with_memory() -> StateGraph: """Create wellness workflow with memory persistence""" graph = create_wellness_workflow() - memory = MemorySaver() + memory = get_checkpointer() return graph.compile(checkpointer=memory) diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..b5fee15 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,4 @@ +[pytest] +pythonpath = . +testpaths = tests +asyncio_mode = auto diff --git a/repos/data_repo.py b/repos/data_repo.py index 9bb77bf..025abff 100644 --- a/repos/data_repo.py +++ b/repos/data_repo.py @@ -1,11 +1,17 @@ from __future__ import annotations import json +import logging +import re import uuid from typing import Any, Dict, List, Optional from pathlib import Path -from datetime import datetime +from datetime import datetime, timedelta, timezone from config.settings import SETTINGS +from repos.graph_auth import GraphAuth, GraphAuthError +from repos.graph_client import GraphAPIError, GraphClient + +logger = logging.getLogger(__name__) def _load_json(path: Path) -> Any: if not path.exists(): return [] if path.suffix == ".json" else None @@ -19,25 +25,146 @@ def _save_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, default=str), encoding="utf-8") + +# ============================================================ +# Microsoft Graph field mapping +# ============================================================ +# Mail and calendar are live Graph data now, not mock JSON. These functions +# translate Graph's field names into the shapes agents/backend already expect +# (see backend/models.py), so nothing downstream has to change. + +_TAG_RE = re.compile(r"<[^>]+>") +_WS_RE = re.compile(r"[ \t]+") + + +def _html_to_text(html: str) -> str: + """Minimal HTML-to-text for email bodies. + + Graph returns HTML-formatted bodies by default. This is a plain tag strip, + not a real HTML parser -- sufficient for an LLM prompt, which does not + care about formatting, and avoids adding a parsing dependency for + something the agents only ever read as flat text. + """ + if not html: + return "" + text = _TAG_RE.sub(" ", html) + text = text.replace(" ", " ").replace("&", "&").replace("<", "<").replace(">", ">") + text = _WS_RE.sub(" ", text) + return "\n".join(line.strip() for line in text.splitlines()).strip() + + +def _map_message(msg: Dict[str, Any]) -> Dict[str, Any]: + """Graph message -> Opspilot's Email shape (backend/models.py:Email).""" + frm = (msg.get("from") or {}).get("emailAddress") or {} + to = [ + r.get("emailAddress", {}).get("address") + for r in (msg.get("toRecipients") or []) + if r.get("emailAddress", {}).get("address") + ] + body = (msg.get("body") or {}).get("content") or msg.get("bodyPreview") or "" + is_html = (msg.get("body") or {}).get("contentType", "").lower() == "html" + + return { + "email_id": msg["id"], + "thread_id": msg.get("conversationId"), + "from_email": frm.get("address", ""), + "sender_name": frm.get("name"), + "to_emails": to, + "subject": msg.get("subject", ""), + "body_text": _html_to_text(body) if is_html else body, + "received_utc": msg.get("receivedDateTime"), + # actionability_gt is a human-labelled ground truth for eval; a live + # mailbox has no such label, so it is always None here by design. + "actionability_gt": None, + } + + +def _map_event(evt: Dict[str, Any]) -> Dict[str, Any]: + """Graph event -> Opspilot's meeting shape (meeting_agent.py).""" + organizer = (evt.get("organizer") or {}).get("emailAddress") or {} + attendees = [ + a.get("emailAddress", {}).get("address") + for a in (evt.get("attendees") or []) + if a.get("emailAddress", {}).get("address") + ] + + join_url = None + if evt.get("isOnlineMeeting"): + join_url = (evt.get("onlineMeeting") or {}).get("joinUrl") + + return { + "meeting_id": evt["id"], + "title": evt.get("subject", ""), + "organizer_email": organizer.get("address", ""), + "participant_emails": attendees, + "scheduled_start_utc": (evt.get("start") or {}).get("dateTime"), + "scheduled_end_utc": (evt.get("end") or {}).get("dateTime"), + "location": (evt.get("location") or {}).get("displayName"), + "agenda": evt.get("bodyPreview"), + # Reused, not repurposed: "transcript_file" has always meant "the + # identifier get_transcript() needs to fetch this meeting's + # transcript." That identifier is now a Teams join URL rather than a + # filename -- meeting_agent.py's call site + # (`get_transcript(mtg.get("transcript_file"))`) does not change. + "transcript_file": join_url, + "correlation_id": None, + } + + +def _run_sync(coro): + """Run an async call from sync code, without assuming there is no running + loop already. + + A plain `asyncio.run` breaks the moment this is called from inside an + already-running loop (e.g. a FastAPI async route awaiting a sync agent + method that turns around and does this). A dedicated thread with its own + loop handles that nested case -- at the cost of a thread hop, which is the + honest price of bridging today's sync call sites onto an async Graph + client rather than making all 31 of them async in the same pass. + """ + import asyncio + import concurrent.futures + + try: + asyncio.get_running_loop() + except RuntimeError: + return asyncio.run(coro) + + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: + return pool.submit(lambda: asyncio.run(coro)).result() + + class DataRepo: + """Mail and calendar: live Microsoft Graph data. Everything else -- + tasks, follow-ups, EOD/weekly reports, governance audit/usage logs -- + Opspilot's own generated output, persisted as JSON. + + That split is deliberate, not partial migration. Graph does not have a + concept of "task priority board" or "minutes of meeting" -- those are + things this system produces by processing real mail and calendar data, + and they need to live somewhere regardless of where the input comes from. + """ + def __init__(self): d = SETTINGS["data"] self.paths = { "users": d["users"], - "inbox": d["emails"]["inbox"], - "threads": d["emails"]["threads"], "tasks": d["tasks"], - "meetings": d["calendar"]["meetings"], "mom": d["calendar"]["mom"], - "transcripts_dir": d["calendar"]["transcripts_dir"], "followups": d["nudges"], "eod": d["reporting"]["eod"], "weekly": d["reporting"]["weekly"], "audit_log": d["governance"]["audit_log"], "llm_usage": d["governance"]["llm_usage"], + # Local processing-state ledger for real emails, keyed by + # email_id -- e.g. {"AAMk...": {"processed": true, ...}}. + # NOT mock mail content: Graph has no concept of "has our agent + # triaged this yet," so that state has to live somewhere, and + # this is Opspilot's own record of it, not a stand-in for real mail. + "email_state": d["emails"]["state"], } - # Cache self._cache = {} + self._graph_client: Optional[GraphClient] = None def _get(self, key: str): if key in self._cache: return self._cache[key] @@ -46,274 +173,300 @@ def _get(self, key: str): self._cache[key] = data return data - # Users + # -- Graph client lifecycle ------------------------------------------- + + def _get_graph_client(self) -> GraphClient: + if self._graph_client is not None: + return self._graph_client + graph_cfg = SETTINGS["graph"] + if not graph_cfg["client_id"]: + raise GraphAuthError( + "Mail and calendar are backed by Microsoft Graph, and " + "MS_GRAPH_CLIENT_ID is not set. See docs/graph_setup.md to " + "create an app registration, then set MS_GRAPH_CLIENT_ID and " + "run scripts/graph_login.py once to complete the device-code login." + ) + auth = GraphAuth( + client_id=graph_cfg["client_id"], + tenant=graph_cfg["tenant"], + cache_path=graph_cfg["token_cache_path"], + ) + self._graph_client = GraphClient(auth) + return self._graph_client + + # -- Users (Opspilot's own directory, not Graph) ----------------------- + def users(self) -> List[Dict[str, Any]]: return self._get("users") def user_by_email(self, email: str) -> Optional[Dict[str, Any]]: return next((u for u in self.users() if u.get("email") == email), None) - # Emails - def inbox(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: - items = self._get("inbox") - return self._apply_filters(items, filters or {}) + # -- Emails: live Graph data + local processing-state merge ----------- - # Tasks - def tasks(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: - items = self._get("tasks") - return self._apply_filters(items, filters or {}) + def _email_state(self) -> Dict[str, Dict[str, Any]]: + raw = _load_json(self.paths["email_state"]) or {} + return raw if isinstance(raw, dict) else {} - # Meetings & Transcripts & MoM - def meetings(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: - items = self._get("meetings") - return self._apply_filters(items, filters or {}) - - def get_transcript(self, transcript_file: Optional[str]) -> str: - if not transcript_file or not isinstance(transcript_file, str): - return "" - path = self.paths["transcripts_dir"] / transcript_file - try: - return path.read_text(encoding="utf-8") - except Exception: - return "" - - def mom_entries(self) -> List[Dict[str, Any]]: - mom = self._get("mom") - return mom if isinstance(mom, list) else [] - - # Followups & Reporting - def followups(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: - return self._apply_filters(self._get("followups"), filters or {}) - - def eod(self) -> List[Dict[str, Any]]: - return self._get("eod") - - def weekly(self) -> List[Dict[str, Any]]: - return self._get("weekly") - - @staticmethod - def _apply_filters(items: List[Dict[str, Any]], filters: Dict[str, Any]) -> List[Dict[str, Any]]: - def match(it): - for k, v in filters.items(): - if v is None: continue - if k not in it: return False - if isinstance(v, (list, tuple, set)): - if it[k] not in v: return False - else: - if it[k] != v: return False - return True - return [it for it in items if match(it)] - - # ============================================================ - # WRITE OPERATIONS (for Agentic AI) - # ============================================================ - - def invalidate_cache(self, key: Optional[str] = None) -> None: - """Clear cache to force reload from disk""" - if key: - self._cache.pop(key, None) - else: - self._cache.clear() + def inbox(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + """Live inbox messages, with local processing-state merged in. + + The mail content (subject, body, sender) always comes from Graph. + `processed` / `agent_actions` / `agent_category` come from this + system's own ledger, keyed by email_id, because Graph has no concept + of those fields. + """ + client = self._get_graph_client() + raw = _run_sync(client.list_messages(top=50)) + state = self._email_state() + + items = [] + for msg in raw: + mapped = _map_message(msg) + saved = state.get(mapped["email_id"], {}) + mapped["processed"] = saved.get("processed", False) + mapped["processed_utc"] = saved.get("processed_utc") + mapped["agent_actions"] = saved.get("agent_actions") + mapped["agent_category"] = saved.get("agent_category") + items.append(mapped) - # --- Email Operations --- - - def add_email_to_inbox(self, email: Dict[str, Any]) -> Dict[str, Any]: - """Add a new email to inbox (for demo sender portal)""" - emails = _load_json(self.paths["inbox"]) or [] - - # Ensure required fields - if "email_id" not in email: - email["email_id"] = f"eml_{uuid.uuid4().hex[:12]}" - if "received_utc" not in email: - email["received_utc"] = datetime.utcnow().isoformat() - if "processed" not in email: - email["processed"] = False - if "agent_actions" not in email: - email["agent_actions"] = [] - - emails.insert(0, email) # Add to top - _save_json(self.paths["inbox"], emails) - self.invalidate_cache("inbox") - return email + return self._apply_filters(items, filters or {}) def get_unprocessed_emails(self) -> List[Dict[str, Any]]: - """Get emails that haven't been processed by agent yet""" - self.invalidate_cache("inbox") # Always fresh - emails = self.inbox() - return [e for e in emails if not e.get("processed", False)] + """Emails the agent hasn't triaged yet, per the local ledger.""" + return [e for e in self.inbox() if not e.get("processed", False)] def mark_email_processed(self, email_id: str, actions_taken: List[str], category: str) -> bool: - """Mark an email as processed by the agent""" - emails = _load_json(self.paths["inbox"]) or [] - - for email in emails: - if email.get("email_id") == email_id: - email["processed"] = True - email["processed_utc"] = datetime.utcnow().isoformat() - email["agent_actions"] = actions_taken - email["agent_category"] = category - _save_json(self.paths["inbox"], emails) - self.invalidate_cache("inbox") - return True - return False + """Record that the agent triaged a real email. + + Writes to the local ledger, not to the mailbox -- Opspilot does not + (and should not) mutate someone's real mail item to record its own + processing state. + """ + state = self._email_state() + state[email_id] = { + "processed": True, + "processed_utc": datetime.now(timezone.utc).isoformat(), + "agent_actions": actions_taken, + "agent_category": category, + } + _save_json(self.paths["email_state"], state) + return True def update_email(self, email_id: str, updates: Dict[str, Any]) -> bool: - """Update email fields""" - emails = _load_json(self.paths["inbox"]) or [] - - for email in emails: - if email.get("email_id") == email_id: - email.update(updates) - email["updated_utc"] = datetime.utcnow().isoformat() - _save_json(self.paths["inbox"], emails) - self.invalidate_cache("inbox") - return True - return False + """Update locally-tracked fields for a real email (ledger, not mailbox).""" + state = self._email_state() + entry = state.get(email_id, {}) + entry.update(updates) + entry["updated_utc"] = datetime.now(timezone.utc).isoformat() + state[email_id] = entry + _save_json(self.paths["email_state"], state) + return True + + # -- Tasks (Opspilot's own derived artifact) --------------------------- + + def tasks(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + items = self._get("tasks") + return self._apply_filters(items, filters or {}) - # --- Task Operations --- - def create_task(self, task: Dict[str, Any]) -> Dict[str, Any]: - """Create a new task""" tasks = _load_json(self.paths["tasks"]) or [] - - # Ensure required fields if "task_id" not in task: task["task_id"] = f"tsk_{uuid.uuid4().hex[:8]}" if "created_utc" not in task: - task["created_utc"] = datetime.utcnow().isoformat() + task["created_utc"] = datetime.now(timezone.utc).isoformat() if "status" not in task: task["status"] = "todo" - tasks.append(task) _save_json(self.paths["tasks"], tasks) self.invalidate_cache("tasks") return task def update_task(self, task_id: str, updates: Dict[str, Any]) -> bool: - """Update an existing task""" tasks = _load_json(self.paths["tasks"]) or [] - for task in tasks: if task.get("task_id") == task_id: task.update(updates) - task["updated_utc"] = datetime.utcnow().isoformat() + task["updated_utc"] = datetime.now(timezone.utc).isoformat() _save_json(self.paths["tasks"], tasks) self.invalidate_cache("tasks") return True return False def delete_task(self, task_id: str) -> bool: - """Delete a task""" tasks = _load_json(self.paths["tasks"]) or [] original_len = len(tasks) tasks = [t for t in tasks if t.get("task_id") != task_id] - if len(tasks) < original_len: _save_json(self.paths["tasks"], tasks) self.invalidate_cache("tasks") return True return False - # --- Follow-up Operations --- - + # -- Meetings: live Graph data, with real event creation --------------- + + def meetings(self, filters: Optional[Dict[str, Any]] = None, *, days_back: int = 14, days_forward: int = 14) -> List[Dict[str, Any]]: + client = self._get_graph_client() + now = datetime.now(timezone.utc) + start = (now - timedelta(days=days_back)).strftime("%Y-%m-%dT%H:%M:%S") + end = (now + timedelta(days=days_forward)).strftime("%Y-%m-%dT%H:%M:%S") + raw = _run_sync(client.list_calendar_view(start_iso=start, end_iso=end)) + items = [_map_event(e) for e in raw] + return self._apply_filters(items, filters or {}) + + def get_transcript(self, transcript_file: Optional[str]) -> str: + """`transcript_file` is the Teams join URL stashed there by + _map_event, not a filename -- see that function's comment.""" + if not transcript_file or not isinstance(transcript_file, str): + return "" + client = self._get_graph_client() + try: + return _run_sync(client.get_transcript_text(transcript_file)) + except GraphAPIError as e: + if e.status_code in (401, 403): + # Expected, named limitation -- not a bug. Personal Microsoft + # accounts cannot access OnlineMeetingTranscript.Read.All under + # any consent; on a work/school tenant it means the admin has + # not granted it. Logged plainly so the gap is visible rather + # than silently swallowed; meeting_agent.py's existing + # no-transcript fallback (a summary from title + attendees) + # takes over from here. + logger.warning( + "Transcript unavailable (HTTP %s) -- expected on a personal " + "Microsoft account; on a work/school tenant this means " + "OnlineMeetingTranscript.Read.All has not been granted. " + "Falling back to calendar-metadata-only MoM generation.", + e.status_code, + ) + return "" + raise + + def create_meeting(self, meeting: Dict[str, Any]) -> Dict[str, Any]: + """Schedule a real calendar event via Graph. + + Used by the "schedule_meeting" approved-action path + (governance/approval.py) -- an agent proposing a follow-up meeting, + once a human approves it, should put a real event on the real + calendar, not append to a local fake meetings list. + """ + client = self._get_graph_client() + event_body = { + "subject": meeting.get("title") or meeting.get("subject", "Follow-up"), + "body": {"contentType": "text", "content": meeting.get("agenda", "")}, + "start": {"dateTime": meeting["scheduled_start_utc"], "timeZone": "UTC"}, + "end": {"dateTime": meeting["scheduled_end_utc"], "timeZone": "UTC"}, + "attendees": [ + {"emailAddress": {"address": addr}, "type": "required"} + for addr in meeting.get("participant_emails", []) + ], + } + created = _run_sync(client.create_event(event_body)) + return _map_event(created) + + def mom_entries(self) -> List[Dict[str, Any]]: + """Previously generated MoM records -- Opspilot's own output, not + something Graph provides (Graph has raw transcripts, not minutes).""" + mom = self._get("mom") + return mom if isinstance(mom, list) else [] + + # -- Follow-ups & reporting (Opspilot's own generated state) ----------- + + def followups(self, filters: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]: + return self._apply_filters(self._get("followups"), filters or {}) + def get_followups(self, status: Optional[str] = None) -> List[Dict[str, Any]]: - """Get all follow-ups, optionally filtered by status""" followups = _load_json(self.paths["followups"]) or [] if status: followups = [f for f in followups if f.get("status") == status] return followups - + def create_followup(self, followup: Dict[str, Any]) -> Dict[str, Any]: - """Create a new follow-up""" followups = _load_json(self.paths["followups"]) or [] - if "followup_id" not in followup: followup["followup_id"] = f"fu_{uuid.uuid4().hex[:8]}" if "created_utc" not in followup: - followup["created_utc"] = datetime.utcnow().isoformat() + followup["created_utc"] = datetime.now(timezone.utc).isoformat() if "status" not in followup: followup["status"] = "pending" - followups.append(followup) _save_json(self.paths["followups"], followups) self.invalidate_cache("followups") return followup def update_followup(self, followup_id: str, updates: Dict[str, Any]) -> bool: - """Update a follow-up""" followups = _load_json(self.paths["followups"]) or [] - for fu in followups: if fu.get("followup_id") == followup_id: fu.update(updates) - fu["updated_utc"] = datetime.utcnow().isoformat() + fu["updated_utc"] = datetime.now(timezone.utc).isoformat() _save_json(self.paths["followups"], followups) self.invalidate_cache("followups") return True return False - # --- Meeting Operations --- - - def create_meeting(self, meeting: Dict[str, Any]) -> Dict[str, Any]: - """Create a new meeting""" - meetings = _load_json(self.paths["meetings"]) or [] - - if "meeting_id" not in meeting: - meeting["meeting_id"] = f"mtg_{uuid.uuid4().hex[:8]}" - if "created_utc" not in meeting: - meeting["created_utc"] = datetime.utcnow().isoformat() - - meetings.append(meeting) - _save_json(self.paths["meetings"], meetings) - self.invalidate_cache("meetings") - return meeting - - # --- Draft Storage --- - + def eod(self) -> List[Dict[str, Any]]: + return self._get("eod") + + def weekly(self) -> List[Dict[str, Any]]: + return self._get("weekly") + + # -- Drafts (agent-generated content awaiting human approval) --------- + def save_draft(self, draft_type: str, draft: Dict[str, Any]) -> Dict[str, Any]: - """Save a draft (email reply, etc.) for review""" drafts_path = self.paths.get("drafts") if not drafts_path: - # Create drafts path if not configured - drafts_path = Path(SETTINGS["data"]["emails"]["inbox"]).parent / "drafts.json" + drafts_path = self.paths["email_state"].parent / "drafts.json" self.paths["drafts"] = drafts_path - drafts = _load_json(drafts_path) or [] - if "draft_id" not in draft: draft["draft_id"] = f"draft_{uuid.uuid4().hex[:8]}" draft["draft_type"] = draft_type - draft["created_utc"] = datetime.utcnow().isoformat() + draft["created_utc"] = datetime.now(timezone.utc).isoformat() draft["status"] = "pending_review" - drafts.append(draft) _save_json(drafts_path, drafts) return draft def get_drafts(self, status: Optional[str] = None) -> List[Dict[str, Any]]: - """Get drafts, optionally filtered by status""" drafts_path = self.paths.get("drafts") if not drafts_path: - drafts_path = Path(SETTINGS["data"]["emails"]["inbox"]).parent / "drafts.json" + drafts_path = self.paths["email_state"].parent / "drafts.json" self.paths["drafts"] = drafts_path - drafts = _load_json(drafts_path) or [] if status: drafts = [d for d in drafts if d.get("status") == status] return drafts def update_draft(self, draft_id: str, updates: Dict[str, Any]) -> bool: - """Update a draft""" drafts_path = self.paths.get("drafts") if not drafts_path: return False - drafts = _load_json(drafts_path) or [] for draft in drafts: if draft.get("draft_id") == draft_id: draft.update(updates) - draft["updated_utc"] = datetime.utcnow().isoformat() + draft["updated_utc"] = datetime.now(timezone.utc).isoformat() _save_json(drafts_path, drafts) return True return False + # -- Shared ------------------------------------------------------------- + + @staticmethod + def _apply_filters(items: List[Dict[str, Any]], filters: Dict[str, Any]) -> List[Dict[str, Any]]: + def match(it): + for k, v in filters.items(): + if v is None: continue + if k not in it: return False + if isinstance(v, (list, tuple, set)): + if it[k] not in v: return False + else: + if it[k] != v: return False + return True + return [it for it in items if match(it)] + + def invalidate_cache(self, key: Optional[str] = None) -> None: + if key: + self._cache.pop(key, None) + else: + self._cache.clear() diff --git a/repos/graph_auth.py b/repos/graph_auth.py new file mode 100644 index 0000000..3cbe33b --- /dev/null +++ b/repos/graph_auth.py @@ -0,0 +1,164 @@ +"""Microsoft Graph authentication: device-code flow. + +Deliberately generic. Nothing here is hardcoded to a specific tenant, app, or +account type -- every value (client_id, tenant, scopes) comes from config, so +the same code authenticates a personal Outlook.com account or a work/school +tenant account, whichever the deployment is configured for. + +Uses MSAL directly rather than the full msgraph-sdk. Opspilot only needs a +handful of Graph endpoints (mail, calendar, transcripts) -- pulling in the +kiota-based SDK for that is a much heavier dependency than the problem needs. +MSAL is Microsoft's own, minimal, and well-maintained auth library; the actual +REST calls go through graph_client.py via plain httpx. + +Why device-code flow specifically: this is a backend service, not a browser +app, so there's no redirect URI to complete an authorization-code exchange +against. Device code lets a user authenticate on a *separate* device (or the +same one) by visiting a short URL and entering a code -- no client secret, +which matters because a public client (this) should never hold one. + +Personal Microsoft accounts and work/school tenant accounts both support +device-code flow through the same app registration, as long as that +registration is configured to allow both account types (see docs/graph_setup.md +for the exact Azure Portal steps -- this is a one-time setup only you can do, +since it requires signing into the Azure Portal). +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import List, Optional + +import msal + +logger = logging.getLogger(__name__) + +# Default scopes: Mail and Calendar read access, which is all the current +# agents need as input. OnlineMeetingTranscript.Read.All is included because +# work/school tenants can grant it -- personal accounts simply won't be able +# to consent to it, and Microsoft's own consent screen handles that +# gracefully (it's just not offered/usable for that account type). Nothing +# breaks by requesting it; see docs/graph_setup.md for what each scope needs. +DEFAULT_SCOPES = [ + "Mail.Read", + "Calendars.Read", + "OnlineMeetingTranscript.Read.All", + "offline_access", +] + + +class GraphAuthError(RuntimeError): + """Authentication could not be completed.""" + + +class GraphAuth: + """Device-code authentication against Microsoft Graph. + + Caches tokens to disk (encrypted-at-rest is the OS's job, not this code's -- + the cache file itself has no special protection beyond filesystem + permissions, so keep it out of anything that gets committed or shipped + elsewhere). A silent refresh is attempted first on every call; the + interactive device-code prompt only fires when there is no usable cached + token, which in practice means once, the first time. + """ + + def __init__( + self, + *, + client_id: str, + tenant: str = "common", + scopes: Optional[List[str]] = None, + cache_path: str | Path = "./.graph_token_cache.json", + ) -> None: + if not client_id: + raise GraphAuthError( + "MS_GRAPH_CLIENT_ID is not set. This must be an app registration " + "you create yourself in the Azure Portal -- see docs/graph_setup.md. " + "There is no default; a shared client_id would let every deployment " + "of this code impersonate every other one." + ) + + self.client_id = client_id + # "common" allows both personal Microsoft accounts and work/school + # tenant accounts to sign in through the same app registration -- this + # is what makes dual account-type support possible without maintaining + # two separate app registrations. + self.authority = f"https://login.microsoftonline.com/{tenant}" + self.scopes = scopes or list(DEFAULT_SCOPES) + self.cache_path = Path(cache_path) + + self._cache = msal.SerializableTokenCache() + if self.cache_path.exists(): + self._cache.deserialize(self.cache_path.read_text(encoding="utf-8")) + + self._app = msal.PublicClientApplication( + client_id=self.client_id, + authority=self.authority, + token_cache=self._cache, + ) + + def _persist_cache(self) -> None: + if self._cache.has_state_changed: + self.cache_path.write_text(self._cache.serialize(), encoding="utf-8") + + def get_token(self, *, interactive: bool = True) -> str: + """Return a valid access token, refreshing or prompting as needed. + + Silent-first: MSAL checks the cache and transparently refreshes an + expired token using the cached refresh token, with no user + interaction. The device-code prompt is the fallback for when there is + nothing usable cached yet. + + `interactive=False` is for callers (like batch/test contexts) that must + not block on a login prompt -- they get GraphAuthError instead of a + hang. + """ + accounts = self._app.get_accounts() + result = None + if accounts: + result = self._app.acquire_token_silent(self.scopes, account=accounts[0]) + + if not result: + if not interactive: + raise GraphAuthError( + "No cached Graph token and interactive=False. " + "Run scripts/graph_login.py once to complete the device-code login." + ) + result = self._device_code_login() + + self._persist_cache() + + if "access_token" not in result: + raise GraphAuthError( + f"Graph authentication failed: {result.get('error')}: " + f"{result.get('error_description')}" + ) + return result["access_token"] + + def _device_code_login(self) -> dict: + flow = self._app.initiate_device_flow(scopes=self.scopes) + if "user_code" not in flow: + raise GraphAuthError(f"Failed to start device-code flow: {flow}") + + # This message is the entire point of device-code flow: the user reads + # it and completes the login themselves, in a real browser, on any + # device. Nothing here can complete that step programmatically -- it + # is Microsoft's consent screen, not this application's. + print(flow["message"]) + logger.info("Graph device-code login started; waiting for user to complete it") + + result = self._app.acquire_token_by_device_flow(flow) + return result + + def forget(self) -> None: + """Remove all cached accounts and delete the on-disk cache. + + Use this to switch which Microsoft account is signed in, or to revoke + local access without touching anything on the Microsoft side. + """ + for account in self._app.get_accounts(): + self._app.remove_account(account) + self._persist_cache() + if self.cache_path.exists(): + self.cache_path.unlink() diff --git a/repos/graph_client.py b/repos/graph_client.py new file mode 100644 index 0000000..793c61a --- /dev/null +++ b/repos/graph_client.py @@ -0,0 +1,210 @@ +"""Thin async client for the Microsoft Graph REST API. + +Deliberately not the msgraph-sdk. This project touches four endpoints -- +list mail, get calendar events, look up an online meeting, and fetch its +transcript. The official SDK is a kiota-generated client covering the entire +Graph surface; pulling in that dependency tree for four endpoints is a much +larger maintenance and install-size cost than writing the four calls directly +against the REST API with httpx. + +Every call goes through `_get`, which raises GraphAPIError with the HTTP +status attached, so callers (graph_repo.py) can distinguish "this account +genuinely cannot do this" (403) from "something is actually broken" (500, +network failure) rather than treating every failure the same way. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +import httpx + +from repos.graph_auth import GraphAuth + +logger = logging.getLogger(__name__) + +GRAPH_BASE = "https://graph.microsoft.com/v1.0" + + +class GraphAPIError(RuntimeError): + """A Graph API call failed. `.status_code` distinguishes the failure kind.""" + + def __init__(self, message: str, *, status_code: Optional[int] = None): + super().__init__(message) + self.status_code = status_code + + +class GraphClient: + """Async wrapper over the Graph REST endpoints Opspilot actually uses.""" + + def __init__(self, auth: GraphAuth, *, timeout: float = 30.0): + self.auth = auth + self._client = httpx.AsyncClient(timeout=timeout) + + async def aclose(self) -> None: + await self._client.aclose() + + async def _headers(self, *, prefer_utc: bool = False) -> Dict[str, str]: + token = self.auth.get_token() + headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"} + if prefer_utc: + # Without this, start/end come back in whatever timezone the + # organizer's mailbox is set to, and converting that correctly + # needs a timezone database dependency this project otherwise has + # no reason to carry. Asking Graph to do the conversion is simpler + # and cannot be wrong the way a manual tz table lookup can. + headers['Prefer'] = 'outlook.timezone="UTC"' + return headers + + async def _get( + self, + url: str, + *, + params: Optional[Dict[str, Any]] = None, + prefer_utc: bool = False, + ) -> Dict[str, Any]: + headers = await self._headers(prefer_utc=prefer_utc) + resp = await self._client.get(url, headers=headers, params=params) + return self._unwrap(resp, "GET", url) + + async def _post(self, url: str, *, json_body: Dict[str, Any]) -> Dict[str, Any]: + headers = await self._headers() + resp = await self._client.post(url, headers=headers, json=json_body) + return self._unwrap(resp, "POST", url) + + @staticmethod + def _unwrap(resp: httpx.Response, method: str, url: str) -> Dict[str, Any]: + if resp.status_code >= 400: + detail = "" + try: + detail = resp.json().get("error", {}).get("message", "") + except Exception: + detail = resp.text[:300] + raise GraphAPIError( + f"{method} {url} -> {resp.status_code}: {detail}", + status_code=resp.status_code, + ) + return resp.json() if resp.content else {} + + async def _get_paged( + self, + url: str, + *, + params: Optional[Dict[str, Any]] = None, + max_pages: int = 10, + prefer_utc: bool = False, + ) -> List[Dict[str, Any]]: + """Follow @odata.nextLink up to `max_pages`. + + Capped rather than unbounded: an unbounded loop against a live mailbox + is a real cost and latency risk, not a theoretical one. + """ + items: List[Dict[str, Any]] = [] + next_url = url + next_params = params + for _ in range(max_pages): + page = await self._get(next_url, params=next_params, prefer_utc=prefer_utc) + items.extend(page.get("value", [])) + next_url = page.get("@odata.nextLink") + next_params = None # nextLink already encodes all query params + if not next_url: + break + return items + + # -- Mail ----------------------------------------------------------- + + async def list_messages( + self, *, top: int = 25, folder: str = "inbox" + ) -> List[Dict[str, Any]]: + """Raw Graph message objects from the given mail folder.""" + url = f"{GRAPH_BASE}/me/mailFolders/{folder}/messages" + params = { + "$top": top, + "$orderby": "receivedDateTime desc", + "$select": ( + "id,conversationId,subject,bodyPreview,body,from,toRecipients," + "receivedDateTime,isRead" + ), + } + return await self._get_paged(url, params=params, max_pages=4) + + async def get_message(self, message_id: str) -> Dict[str, Any]: + return await self._get(f"{GRAPH_BASE}/me/messages/{message_id}") + + # -- Calendar --------------------------------------------------------- + + async def list_calendar_view( + self, *, start_iso: str, end_iso: str + ) -> List[Dict[str, Any]]: + """Raw Graph event objects in [start_iso, end_iso). + + calendarView (not /events) is deliberate: it expands recurring + meetings into their actual occurrences in the window, which is what a + "what's on my calendar this week" read needs. /events would return the + recurrence master only. + """ + url = f"{GRAPH_BASE}/me/calendarView" + params = { + "startDateTime": start_iso, + "endDateTime": end_iso, + "$orderby": "start/dateTime", + "$select": ( + "id,subject,organizer,attendees,start,end,location,bodyPreview," + "onlineMeeting,isOnlineMeeting" + ), + } + return await self._get_paged(url, params=params, max_pages=4, prefer_utc=True) + + async def create_event(self, event: Dict[str, Any]) -> Dict[str, Any]: + """Create a real calendar event. + + `event` must already be in Graph's event shape (subject, start, end, + attendees, ...). This is a deliberate boundary: the caller (DataRepo) + is responsible for building that shape from Opspilot's internal + "schedule_meeting" action payload, so this client stays a thin, + honest wrapper over the Graph endpoint rather than absorbing + Opspilot-specific field-naming knowledge. + """ + return await self._post(f"{GRAPH_BASE}/me/events", json_body=event) + + # -- Online meeting transcripts --------------------------------------- + + async def get_transcript_text(self, join_web_url: str) -> str: + """Best-effort transcript fetch for a Teams meeting. + + Returns "" (not an exception) for the expected failure cases: no + transcript exists yet, or the account type cannot access transcripts + at all (see graph_repo.py, which is where that distinction is logged + clearly -- this method only raises for genuinely unexpected failures). + """ + meetings = await self._get( + f"{GRAPH_BASE}/me/onlineMeetings", + params={"$filter": f"JoinWebUrl eq '{join_web_url}'"}, + ) + values = meetings.get("value", []) + if not values: + return "" + meeting_id = values[0]["id"] + + transcripts = await self._get( + f"{GRAPH_BASE}/me/onlineMeetings/{meeting_id}/transcripts" + ) + entries = transcripts.get("value", []) + if not entries: + return "" + + transcript_id = entries[0]["id"] + headers = await self._headers() + headers["Accept"] = "text/vtt" + resp = await self._client.get( + f"{GRAPH_BASE}/me/onlineMeetings/{meeting_id}/transcripts/{transcript_id}/content", + headers=headers, + params={"$format": "text/vtt"}, + ) + if resp.status_code >= 400: + raise GraphAPIError( + f"transcript content fetch -> {resp.status_code}", + status_code=resp.status_code, + ) + return resp.text diff --git a/requirements.txt b/requirements.txt index 780ee73..32e49e5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,12 +1,41 @@ -streamlit==1.40.1 -pydantic==2.7.4 +# Core +pydantic>=2.7,<3 python-dotenv==1.0.1 pandas==2.2.2 -litellm==1.42.6 -langgraph==0.2.45 -langchain-core==0.3.20 orjson==3.10.7 rich==13.9.2 -nicegui openai>=1.0.0 + +# Orchestration. langgraph pulls in langchain-core transitively (>=0.1) -- +# not pinned explicitly here because nothing in this codebase imports +# langchain_core directly; only langgraph's own public API (StateGraph, END) +# is used, and that surface is unchanged across the 0.2.x -> 1.x jump. +langgraph>=1.0.0,<2.0.0 + +# Checkpointing. Redis-backed rather than the in-process MemorySaver every +# graph used before -- see orchestration/checkpointer.py for why that +# mattered. redis is the Python client langgraph-checkpoint-redis needs. +langgraph-checkpoint-redis>=1.0.0 +redis>=5.0.0 + +# Microsoft Graph (mail + calendar). msal for device-code auth, httpx for the +# REST calls -- see repos/graph_auth.py and repos/graph_client.py for why this +# is a thin direct integration rather than the full msgraph-sdk. +msal>=1.30.0 +httpx>=0.27.0 + +# Vector memory chromadb>=0.4.0 + +# Note: litellm was pinned here but never imported anywhere in this codebase +# (governance/litellm_gateway.py talks to an OpenAI-API-compatible endpoint +# directly via the `openai` SDK -- the name predates that, kept only for the +# seven modules that already import EnhancedLiteLLMGateway from that path). +# Removed rather than left as dead weight. + +# Note: streamlit and nicegui were pinned here as a supposed "second UI +# stack" (issue #4), but neither is actually imported anywhere in this +# codebase -- `grep -rn "import streamlit\|import nicegui"` across the whole +# tree returns nothing. There is one real UI: the Next.js frontend (9 pages +# under frontend/src/app). These were unused dependency weight, not a second +# implementation to choose between. Removed. diff --git a/scripts/graph_login.py b/scripts/graph_login.py new file mode 100644 index 0000000..60380ef --- /dev/null +++ b/scripts/graph_login.py @@ -0,0 +1,65 @@ +"""Interactive Microsoft Graph login for Opspilot. + + python -m scripts.graph_login # sign in (device-code flow) + python -m scripts.graph_login --logout # clear the cached account + +This is the one step in the whole Graph integration that genuinely requires a +human: Microsoft's consent screen. Nothing in the rest of the codebase can +complete it programmatically, and nothing should try to -- that consent is +the entire point of delegated auth. + +Run this once. After that, GraphAuth refreshes the cached token silently on +every subsequent run; you should only need to run this again if you delete +the token cache or revoke access. +""" + +from __future__ import annotations + +import argparse +import sys + +from config.settings import SETTINGS +from repos.graph_auth import GraphAuth, GraphAuthError + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--logout", action="store_true", + help="Clear the cached account and token, then exit.", + ) + args = parser.parse_args() + + graph_cfg = SETTINGS["graph"] + if not graph_cfg["client_id"]: + print( + "MS_GRAPH_CLIENT_ID is not set. Create an app registration first " + "-- see docs/graph_setup.md -- then set MS_GRAPH_CLIENT_ID in .env.", + file=sys.stderr, + ) + return 1 + + auth = GraphAuth( + client_id=graph_cfg["client_id"], + tenant=graph_cfg["tenant"], + cache_path=graph_cfg["token_cache_path"], + ) + + if args.logout: + auth.forget() + print("Cleared the cached Microsoft Graph account and token.") + return 0 + + print("Starting Microsoft Graph device-code login...\n") + try: + auth.get_token(interactive=True) + except GraphAuthError as e: + print(f"Login failed: {e}", file=sys.stderr) + return 1 + + print("\nSigned in. The token is cached and will refresh automatically on future runs.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/comprehensive_quality_tests.py b/tests/comprehensive_quality_tests.py index fbf1b5c..66a1de2 100644 --- a/tests/comprehensive_quality_tests.py +++ b/tests/comprehensive_quality_tests.py @@ -70,14 +70,14 @@ class QualityAnalyzer: def __init__(self): self.client = None - if HAS_OPENAI and SETTINGS["models"]["azure_api_key"]: + if HAS_OPENAI and SETTINGS["llm"]["api_key"]: try: self.client = AzureOpenAI( - api_key=SETTINGS["models"]["azure_api_key"], + api_key=SETTINGS["llm"]["api_key"], api_version="2024-02-15-preview", - azure_endpoint=SETTINGS["models"]["azure_api_base"] + azure_endpoint=SETTINGS["llm"]["base_url"] ) - self.model = SETTINGS["models"]["chat_model"] + self.model = SETTINGS["llm"]["model_id"] print(f"✓ Azure OpenAI initialized with model: {self.model}") except Exception as e: print(f"✗ Failed to initialize Azure OpenAI: {e}") diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..489b714 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,14 @@ +"""Test configuration. + +No environment variables are strictly required to import config.settings -- +every value has a plain default. Tests that need Graph or Redis configured set +what they need directly, so each test's requirements are visible at its own +call site rather than hidden in a shared fixture. +""" + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) diff --git a/tests/test_app_boots.py b/tests/test_app_boots.py new file mode 100644 index 0000000..6a36783 --- /dev/null +++ b/tests/test_app_boots.py @@ -0,0 +1,81 @@ +"""The FastAPI app must actually import and serve. + +This is the regression test for issue #1 in CLAUDE.md: "nothing here has +been shown to run end-to-end." The specific bug that blocked it -- +backend/routes_ai.py importing `app.smart_chat.SmartChatAgent`, a module +that was never committed anywhere in this repository -- took the entire +FastAPI app down at import time over three endpoints out of several dozen. +""" + +from __future__ import annotations + +import pytest + + +@pytest.fixture(scope="module") +def client(): + from fastapi.testclient import TestClient + import backend.app as app_module + + # raise_server_exceptions=False: some /ai/* endpoints construct an LLM + # gateway eagerly and this test environment has no LLM configured, which + # correctly raises LLMNotConfiguredError (see test_gateway.py -- that's + # the fail-closed behavior working as intended, not a bug). This suite is + # about the SEPARATE SmartChatAgent import regression, so a 500 from an + # unrelated, expected cause should come back as a normal response to + # assert against, not propagate and fail these tests for the wrong reason. + with TestClient(app_module.app, raise_server_exceptions=False) as c: + yield c + + +class TestAppImports: + def test_app_module_imports_without_smart_chat(self): + """The regression this test exists for: importing backend.app must + not require app.smart_chat, which does not exist in this repo.""" + import sys + + import backend.app # noqa: F401 + + assert "app.smart_chat" not in sys.modules + + def test_app_registers_a_substantial_number_of_routes(self): + import backend.app as app_module + + assert len(app_module.app.routes) > 20 + + +class TestUnaffectedEndpointsStillWork: + """The ten /ai/* endpoints that don't touch SmartChatAgent must be + completely unaffected by its absence.""" + + def test_nudges_endpoint_responds(self, client): + r = client.get("/api/v1/ai/nudges") + # Whatever the status, it must not be because the router itself + # failed to load -- that would be a 404 for an unregistered route. + assert r.status_code != 404 + + def test_weekly_reports_endpoint_responds(self, client): + r = client.get("/api/v1/ai/reports/weekly") + assert r.status_code != 404 + + +class TestAssistantEndpointsFailClearly: + """The three endpoints that DO need SmartChatAgent must answer 501 -- + not fabricate a response, not 500, not silently do nothing.""" + + def test_assistant_start_returns_501(self, client): + r = client.post("/api/v1/assistant/start", json={"user_email": "a@example.com"}) + assert r.status_code == 501 + assert "smart_chat" in r.json()["detail"].lower() or "assistant" in r.json()["detail"].lower() + + def test_assistant_chat_returns_501(self, client): + r = client.post( + "/api/v1/assistant/chat", + json={"session_id": "x", "user_email": "a@example.com", "message": "hi"}, + ) + assert r.status_code == 501 + + def test_error_names_the_missing_capability_not_a_generic_failure(self, client): + r = client.post("/api/v1/assistant/start", json={"user_email": "a@example.com"}) + detail = r.json()["detail"] + assert "never implemented" in detail or "not available" in detail diff --git a/tests/test_checkpointer.py b/tests/test_checkpointer.py new file mode 100644 index 0000000..33033e8 --- /dev/null +++ b/tests/test_checkpointer.py @@ -0,0 +1,77 @@ +"""Shared Redis checkpointer: fails clearly when Redis is unreachable, rather +than silently falling back to in-process storage (which would quietly +reintroduce the exact "state lost on restart" problem this module replaces +MemorySaver to fix). +""" + +from __future__ import annotations + +import socket + +import pytest + +import orchestration.checkpointer as checkpointer_module +from orchestration.checkpointer import CheckpointerUnavailable, get_checkpointer + + +@pytest.fixture(autouse=True) +def reset_shared_state(): + """Each test gets a clean slate -- the module caches the constructed + saver process-wide, which would let one test's Redis config leak into + the next.""" + checkpointer_module._saver = None + checkpointer_module._setup_done = False + yield + checkpointer_module._saver = None + checkpointer_module._setup_done = False + + +def _port_is_free(port: int) -> bool: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + return s.connect_ex(("localhost", port)) != 0 + + +class TestUnreachableRedis: + def test_raises_checkpointer_unavailable_with_actionable_message(self, monkeypatch): + import config.settings as settings_module + + # Port 1 is privileged and essentially never has anything listening + # in a test environment -- a reliable "nothing is here" target. + monkeypatch.setitem(settings_module.SETTINGS["redis"], "url", "redis://localhost:1/0") + + with pytest.raises(CheckpointerUnavailable) as exc: + get_checkpointer() + + message = str(exc.value) + assert "localhost:1" in message + assert "docker run" in message # the actionable remediation hint + + def test_does_not_silently_fall_back_to_in_memory_storage(self, monkeypatch): + """The property that matters most: a missing Redis must be loud, not + a quiet reversion to the exact per-process-memory problem this + module exists to replace.""" + import config.settings as settings_module + + monkeypatch.setitem(settings_module.SETTINGS["redis"], "url", "redis://localhost:1/0") + + with pytest.raises(CheckpointerUnavailable): + get_checkpointer() + + # No MemorySaver-shaped object was quietly cached in its place. + assert checkpointer_module._saver is None + + +@pytest.mark.skipif(_port_is_free(6379), reason="no local Redis reachable on 6379") +class TestWithLiveRedis: + """Skipped unless a real Redis is reachable on the default port -- see + docs in orchestration/checkpointer.py for how to run one locally.""" + + def test_returns_the_same_instance_on_repeated_calls(self): + first = get_checkpointer() + second = get_checkpointer() + assert first is second + + def test_setup_runs_only_once_across_calls(self): + get_checkpointer() + assert checkpointer_module._setup_done is True + get_checkpointer() # must not re-run setup or raise diff --git a/tests/test_data_repo.py b/tests/test_data_repo.py new file mode 100644 index 0000000..ca40ba1 --- /dev/null +++ b/tests/test_data_repo.py @@ -0,0 +1,176 @@ +"""DataRepo: Graph field mapping, local processing-state merge, fail-closed +behaviour when Graph isn't configured. + +The mapping functions (_map_message, _map_event, _html_to_text) are pure -- +tested directly with hand-built Graph-shaped payloads, not live data. This is +the boundary: what Graph actually returns is real and live; the payload shape +used to exercise the mapping code is a fixture for the mapping code, not a +stand-in for real mail or calendar content anywhere the app runs. +""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest + +from repos.data_repo import DataRepo, _html_to_text, _map_event, _map_message +from repos.graph_auth import GraphAuthError + + +class TestHtmlToText: + def test_strips_tags(self): + assert _html_to_text("

Hello world

") == "Hello world" + + def test_decodes_common_entities(self): + assert _html_to_text("Tom & Jerry  here") == "Tom & Jerry here" + + def test_empty_input(self): + assert _html_to_text("") == "" + assert _html_to_text(None) == "" + + def test_collapses_whitespace_within_a_line(self): + assert _html_to_text("

a b

") == "a b" + + +class TestMapMessage: + def _graph_message(self, **overrides): + base = { + "id": "msg1", + "conversationId": "conv1", + "subject": "Hello", + "from": {"emailAddress": {"address": "a@example.com", "name": "A"}}, + "toRecipients": [{"emailAddress": {"address": "b@example.com"}}], + "body": {"contentType": "text", "content": "plain body"}, + "receivedDateTime": "2026-01-01T00:00:00Z", + } + base.update(overrides) + return base + + def test_maps_core_fields(self): + mapped = _map_message(self._graph_message()) + assert mapped["email_id"] == "msg1" + assert mapped["thread_id"] == "conv1" + assert mapped["from_email"] == "a@example.com" + assert mapped["sender_name"] == "A" + assert mapped["to_emails"] == ["b@example.com"] + assert mapped["subject"] == "Hello" + assert mapped["body_text"] == "plain body" + assert mapped["received_utc"] == "2026-01-01T00:00:00Z" + + def test_html_body_is_converted_to_text(self): + mapped = _map_message(self._graph_message( + body={"contentType": "html", "content": "

Hi there

"} + )) + assert mapped["body_text"] == "Hi there" + + def test_actionability_gt_is_always_none(self): + """A human-labelled eval field; a live mailbox has no such label.""" + mapped = _map_message(self._graph_message()) + assert mapped["actionability_gt"] is None + + def test_missing_from_does_not_crash(self): + mapped = _map_message(self._graph_message(**{"from": None})) + assert mapped["from_email"] == "" + + def test_missing_to_recipients_gives_empty_list(self): + mapped = _map_message(self._graph_message(toRecipients=[])) + assert mapped["to_emails"] == [] + + +class TestMapEvent: + def _graph_event(self, **overrides): + base = { + "id": "evt1", + "subject": "Sync", + "organizer": {"emailAddress": {"address": "org@example.com"}}, + "attendees": [{"emailAddress": {"address": "att@example.com"}}], + "start": {"dateTime": "2026-01-01T10:00:00"}, + "end": {"dateTime": "2026-01-01T11:00:00"}, + "location": {"displayName": "Room 1"}, + "bodyPreview": "Agenda text", + "isOnlineMeeting": False, + } + base.update(overrides) + return base + + def test_maps_core_fields(self): + mapped = _map_event(self._graph_event()) + assert mapped["meeting_id"] == "evt1" + assert mapped["title"] == "Sync" + assert mapped["organizer_email"] == "org@example.com" + assert mapped["participant_emails"] == ["att@example.com"] + assert mapped["scheduled_start_utc"] == "2026-01-01T10:00:00" + assert mapped["scheduled_end_utc"] == "2026-01-01T11:00:00" + assert mapped["location"] == "Room 1" + assert mapped["agenda"] == "Agenda text" + + def test_non_online_meeting_has_no_transcript_ref(self): + mapped = _map_event(self._graph_event(isOnlineMeeting=False)) + assert mapped["transcript_file"] is None + + def test_online_meeting_carries_join_url_as_transcript_ref(self): + """Reused field, not repurposed: transcript_file has always meant + "the identifier get_transcript() needs" -- here that's a join URL.""" + mapped = _map_event(self._graph_event( + isOnlineMeeting=True, + onlineMeeting={"joinUrl": "https://teams.microsoft.com/l/meetup/xyz"}, + )) + assert mapped["transcript_file"] == "https://teams.microsoft.com/l/meetup/xyz" + + +class TestDataRepoFailsClosedWithoutGraphConfig: + def test_inbox_raises_clear_error(self, monkeypatch): + monkeypatch.setenv("MS_GRAPH_CLIENT_ID", "") + import importlib + import config.settings as settings_module + importlib.reload(settings_module) + + repo = DataRepo.__new__(DataRepo) + repo._graph_client = None + repo.paths = {} + with pytest.raises(GraphAuthError, match="MS_GRAPH_CLIENT_ID"): + repo._get_graph_client() + + +class TestEmailProcessingLedger: + """The local ledger (processed/agent_actions/agent_category) is Opspilot's + own state about REAL emails -- it must never be mistaken for mail content + itself, and it must persist independently of whatever Graph returns.""" + + @pytest.fixture + def repo(self, tmp_path, monkeypatch): + import config.settings as settings_module + settings_module.SETTINGS["data"]["emails"]["state"] = tmp_path / "email_state.json" + settings_module.SETTINGS["data"]["tasks"] = tmp_path / "tasks.json" + settings_module.SETTINGS["data"]["nudges"] = tmp_path / "followups.json" + settings_module.SETTINGS["data"]["reporting"]["eod"] = tmp_path / "eod.json" + settings_module.SETTINGS["data"]["reporting"]["weekly"] = tmp_path / "weekly.json" + settings_module.SETTINGS["data"]["calendar"]["mom"] = tmp_path / "mom.json" + settings_module.SETTINGS["data"]["users"] = tmp_path / "users.json" + return DataRepo() + + def test_mark_processed_then_read_back(self, repo): + repo.mark_email_processed("msg1", ["archived"], "noise") + state = repo._email_state() + assert state["msg1"]["processed"] is True + assert state["msg1"]["agent_actions"] == ["archived"] + assert state["msg1"]["agent_category"] == "noise" + + def test_unmarked_email_defaults_to_unprocessed(self, repo): + state = repo._email_state() + assert state.get("never-seen", {}).get("processed", False) is False + + def test_update_email_merges_rather_than_replaces(self, repo): + repo.mark_email_processed("msg1", ["a"], "cat1") + repo.update_email("msg1", {"agent_category": "cat2"}) + state = repo._email_state() + assert state["msg1"]["agent_category"] == "cat2" + assert state["msg1"]["processed"] is True # untouched by the update + + def test_ledger_persists_across_repo_instances(self, repo, tmp_path): + repo.mark_email_processed("msg1", ["x"], "y") + repo2 = DataRepo() + assert repo2._email_state()["msg1"]["processed"] is True diff --git a/tests/test_gateway.py b/tests/test_gateway.py new file mode 100644 index 0000000..8c45ff0 --- /dev/null +++ b/tests/test_gateway.py @@ -0,0 +1,92 @@ +"""LLM gateway: fails closed instead of fabricating. + +This used to have three layers of fake-response fallback -- PolicyGateway's +own simulate, EnhancedLiteLLMGateway's simulate, and PolicyGateway +re-simulating if the gateway raised -- all returning literal placeholder +content like "Decision 1", "Decision 2" when unconfigured. These tests assert +that path is actually gone, not just documented as gone. +""" + +from __future__ import annotations + +import pytest + +from governance.gateway import PolicyGateway +from governance.litellm_gateway import DiskCache, EnhancedLiteLLMGateway, LLMNotConfiguredError + + +@pytest.fixture(autouse=True) +def clear_llm_config(monkeypatch): + """Every test in this file starts from "nothing configured" unless it + sets its own SETTINGS values -- otherwise a previous test's config, or + something picked up from a real .env, would leak in.""" + import config.settings as settings_module + monkeypatch.setitem(settings_module.SETTINGS["llm"], "base_url", "") + monkeypatch.setitem(settings_module.SETTINGS["llm"], "api_key", "") + monkeypatch.setitem(settings_module.SETTINGS["llm"], "model_id", "") + + +class TestFailsClosedWhenUnconfigured: + def test_gateway_raises_at_construction_not_first_call(self): + """Fail at construction, not three steps into a workflow when an + agent finally tries to use it.""" + with pytest.raises(LLMNotConfiguredError): + EnhancedLiteLLMGateway("test_agent") + + def test_policy_gateway_propagates_the_same_error(self): + with pytest.raises(LLMNotConfiguredError): + PolicyGateway("email_agent") + + def test_error_message_names_all_three_required_variables(self): + with pytest.raises(LLMNotConfiguredError, match="LLM_BASE_URL"): + EnhancedLiteLLMGateway("test_agent") + + @pytest.mark.parametrize("missing_key", ["base_url", "api_key", "model_id"]) + def test_any_single_missing_value_is_enough_to_fail(self, missing_key): + import config.settings as settings_module + settings_module.SETTINGS["llm"]["base_url"] = "https://example.com" + settings_module.SETTINGS["llm"]["api_key"] = "key" + settings_module.SETTINGS["llm"]["model_id"] = "model" + settings_module.SETTINGS["llm"][missing_key] = "" + + with pytest.raises(LLMNotConfiguredError): + EnhancedLiteLLMGateway("test_agent") + + +class TestNoFabricatedFallbackRemains: + """The specific methods that used to return canned content are gone + entirely, not merely unreachable.""" + + def test_enhanced_gateway_has_no_simulate_method(self): + assert not hasattr(EnhancedLiteLLMGateway, "_simulate_response") + + def test_policy_gateway_has_no_simulate_method(self): + assert not hasattr(PolicyGateway, "_simulate_response") + assert not hasattr(PolicyGateway, "_generate_dynamic_mom") + + +class TestDiskCache: + def test_roundtrip(self, tmp_path): + cache = DiskCache(cache_dir=str(tmp_path)) + cache.set("prompt", "model-x", 0.2, "the response", 10, 5) + assert cache.get("prompt", "model-x", 0.2) == "the response" + + def test_different_temperature_is_a_cache_miss(self, tmp_path): + cache = DiskCache(cache_dir=str(tmp_path)) + cache.set("prompt", "model-x", 0.2, "response", 10, 5) + assert cache.get("prompt", "model-x", 0.9) is None + + def test_expired_entry_is_a_miss(self, tmp_path): + cache = DiskCache(cache_dir=str(tmp_path), ttl_hours=0) + cache.set("prompt", "model-x", 0.2, "response", 10, 5) + import time + time.sleep(0.01) + assert cache.get("prompt", "model-x", 0.2) is None + + def test_clear_removes_all_entries(self, tmp_path): + cache = DiskCache(cache_dir=str(tmp_path)) + cache.set("p1", "m", 0.2, "r1", 1, 1) + cache.set("p2", "m", 0.2, "r2", 1, 1) + cache.clear() + assert cache.get("p1", "m", 0.2) is None + assert cache.get("p2", "m", 0.2) is None diff --git a/tests/test_graph_client.py b/tests/test_graph_client.py new file mode 100644 index 0000000..f6a16d8 --- /dev/null +++ b/tests/test_graph_client.py @@ -0,0 +1,178 @@ +"""GraphClient: pagination, error handling, transcript fetch. + +The HTTP boundary is stubbed with respx -- this is test-boundary mocking, not +product mock data. The distinction matters: these tests assert GraphClient +correctly interprets whatever the real Graph API would send back, they never +stand in for real mail/calendar content anywhere in the running application. +""" + +from __future__ import annotations + +import pytest +import respx +import httpx + +from repos.graph_client import GRAPH_BASE, GraphAPIError, GraphClient + + +class FakeAuth: + def get_token(self) -> str: + return "fake-token" + + +@pytest.fixture +def client(): + c = GraphClient(FakeAuth()) + yield c + + +class TestErrorUnwrapping: + @pytest.mark.asyncio + async def test_success_returns_json(self, client): + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/messages/abc").mock( + return_value=httpx.Response(200, json={"id": "abc", "subject": "hi"}) + ) + result = await client.get_message("abc") + assert result == {"id": "abc", "subject": "hi"} + + @pytest.mark.asyncio + async def test_403_raises_with_status_code(self, client): + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/messages/x").mock( + return_value=httpx.Response( + 403, json={"error": {"message": "Access denied"}} + ) + ) + with pytest.raises(GraphAPIError) as exc: + await client.get_message("x") + assert exc.value.status_code == 403 + assert "Access denied" in str(exc.value) + + @pytest.mark.asyncio + async def test_error_body_that_is_not_json_does_not_crash(self, client): + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/messages/x").mock( + return_value=httpx.Response(500, text="upstream on fire") + ) + with pytest.raises(GraphAPIError) as exc: + await client.get_message("x") + assert exc.value.status_code == 500 + + +class TestPagination: + @pytest.mark.asyncio + async def test_follows_odata_next_link(self, client): + """A mailbox with more messages than one page returns spans multiple + requests, chained via @odata.nextLink.""" + page2_url = f"{GRAPH_BASE}/me/mailFolders/inbox/messages?page=2" + responses = [ + httpx.Response( + 200, + json={"value": [{"id": "1"}, {"id": "2"}], "@odata.nextLink": page2_url}, + ), + httpx.Response(200, json={"value": [{"id": "3"}]}), + ] + + def responder(request): + return responses.pop(0) + + with respx.mock: + respx.get(url__startswith=f"{GRAPH_BASE}/me/mailFolders/inbox/messages").mock( + side_effect=responder + ) + items = await client.list_messages(top=2) + assert [i["id"] for i in items] == ["1", "2", "3"] + + @pytest.mark.asyncio + async def test_stops_at_max_pages(self, client): + """An unbounded page-follow loop against a live mailbox is a real cost + and latency risk -- this asserts the cap actually bites.""" + call_count = 0 + + def responder(request): + nonlocal call_count + call_count += 1 + return httpx.Response( + 200, + json={ + "value": [{"id": str(call_count)}], + "@odata.nextLink": f"{GRAPH_BASE}/me/mailFolders/inbox/messages?p={call_count}", + }, + ) + + with respx.mock: + respx.get(url__startswith=f"{GRAPH_BASE}/me/mailFolders/inbox/messages").mock( + side_effect=responder + ) + items = await client.list_messages(top=1) + # max_pages=4 in list_messages + assert call_count == 4 + assert len(items) == 4 + + +class TestTranscripts: + @pytest.mark.asyncio + async def test_no_matching_online_meeting_returns_empty(self, client): + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/onlineMeetings").mock( + return_value=httpx.Response(200, json={"value": []}) + ) + text = await client.get_transcript_text("https://teams.microsoft.com/x") + assert text == "" + + @pytest.mark.asyncio + async def test_no_transcript_entries_returns_empty(self, client): + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/onlineMeetings").mock( + return_value=httpx.Response(200, json={"value": [{"id": "mtg1"}]}) + ) + respx.get(f"{GRAPH_BASE}/me/onlineMeetings/mtg1/transcripts").mock( + return_value=httpx.Response(200, json={"value": []}) + ) + text = await client.get_transcript_text("https://teams.microsoft.com/x") + assert text == "" + + @pytest.mark.asyncio + async def test_permission_denied_raises_403(self, client): + """The case that matters: a personal Microsoft account, or a + work/school tenant without admin consent for + OnlineMeetingTranscript.Read.All. GraphClient must surface this as a + 403 -- distinguishing it from a real failure is data_repo.py's job, + not this layer's, but it can only do that if the status code survives.""" + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/onlineMeetings").mock( + return_value=httpx.Response( + 403, json={"error": {"message": "not supported for consumer accounts"}} + ) + ) + with pytest.raises(GraphAPIError) as exc: + await client.get_transcript_text("https://teams.microsoft.com/x") + assert exc.value.status_code == 403 + + @pytest.mark.asyncio + async def test_successful_transcript_fetch(self, client): + with respx.mock: + respx.get(f"{GRAPH_BASE}/me/onlineMeetings").mock( + return_value=httpx.Response(200, json={"value": [{"id": "mtg1"}]}) + ) + respx.get(f"{GRAPH_BASE}/me/onlineMeetings/mtg1/transcripts").mock( + return_value=httpx.Response(200, json={"value": [{"id": "t1"}]}) + ) + respx.get(f"{GRAPH_BASE}/me/onlineMeetings/mtg1/transcripts/t1/content").mock( + return_value=httpx.Response(200, text="WEBVTT\n\nhello world") + ) + text = await client.get_transcript_text("https://teams.microsoft.com/x") + assert "hello world" in text + + +class TestEventCreation: + @pytest.mark.asyncio + async def test_create_event_posts_body_and_returns_response(self, client): + with respx.mock: + route = respx.post(f"{GRAPH_BASE}/me/events").mock( + return_value=httpx.Response(201, json={"id": "evt1", "subject": "Sync"}) + ) + result = await client.create_event({"subject": "Sync"}) + assert result == {"id": "evt1", "subject": "Sync"} + assert route.calls.last.request.method == "POST" From c8a29f25545d6c3bd549e102099a5f8bc80a254b Mon Sep 17 00:00:00 2001 From: kowshikdev Date: Thu, 23 Jul 2026 15:12:42 +0530 Subject: [PATCH 2/3] fix: langgraph-checkpoint-redis version pin was wrong, never resolved Pinned >=1.0.0 based on an inaccurate web search result. PyPI's actual latest is 0.5.1 (verified directly: pip index versions langgraph-checkpoint-redis). CI caught it immediately -- 'ERROR: No matching distribution found for langgraph-checkpoint-redis>=1.0.0' -- rather than it surfacing later. Verified every other new pin in this file (msal, httpx, langgraph, redis) against real PyPI data too; those were correct. Co-Authored-By: Claude Opus 4.8 --- requirements.txt | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 32e49e5..a005bd4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -15,7 +15,13 @@ langgraph>=1.0.0,<2.0.0 # Checkpointing. Redis-backed rather than the in-process MemorySaver every # graph used before -- see orchestration/checkpointer.py for why that # mattered. redis is the Python client langgraph-checkpoint-redis needs. -langgraph-checkpoint-redis>=1.0.0 +# +# >=1.0.0 was wrong and never resolves -- PyPI's actual latest is 0.5.1 +# (verified directly: `pip index versions langgraph-checkpoint-redis`). +# An earlier web search claimed a 1.0.10 release; that was inaccurate, and CI +# caught it (`ERROR: No matching distribution found for +# langgraph-checkpoint-redis>=1.0.0`) rather than it surfacing later. +langgraph-checkpoint-redis>=0.5.0,<1.0.0 redis>=5.0.0 # Microsoft Graph (mail + calendar). msal for device-code auth, httpx for the From 212f4ce5b637712a89b832f6ac4f76fa3e8972d5 Mon Sep 17 00:00:00 2001 From: kowshikdev Date: Thu, 23 Jul 2026 15:29:05 +0530 Subject: [PATCH 3/3] fix: route-count test asserted on a FastAPI-version-dependent implementation detail len(app.routes) failed in CI (17) while passing locally (74) -- reproduced by building a clean venv installing exactly what CI installs (no version pin ceiling on fastapi, so it resolves to the actual latest, 0.139.2, versus an older one already present locally). The real cause: newer FastAPI represents each include_router() call as a single lazy _IncludedRouter wrapper in app.routes rather than flattening it into individual Route objects immediately. All 12 router-include calls in backend/app.py were succeeding correctly in both environments -- the test was counting an internal representation detail that changed between versions, not missing endpoints. Fixed to assert on app.openapi()["paths"] instead, which resolves to 51 real paths in the CI-matching venv regardless of FastAPI's internal route representation -- OpenAPI generation has to fully expand routing to build the schema either way. Verified in a clean venv with the exact same install commands CI runs: 48 pass, 2 skip. Co-Authored-By: Claude Opus 4.8 --- tests/test_app_boots.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/tests/test_app_boots.py b/tests/test_app_boots.py index 6a36783..5160587 100644 --- a/tests/test_app_boots.py +++ b/tests/test_app_boots.py @@ -39,9 +39,21 @@ def test_app_module_imports_without_smart_chat(self): assert "app.smart_chat" not in sys.modules def test_app_registers_a_substantial_number_of_routes(self): + """`len(app.routes)` is not a stable thing to assert on: newer FastAPI + represents each include_router() call as a single lazy + `_IncludedRouter` wrapper in `app.routes` rather than flattening it + into individual Route objects immediately. Counting `app.routes` + directly gave 74 against one locally-installed FastAPI version and 17 + against another (0.139.2, what CI actually installs, unpinned) -- + with every router genuinely registered correctly in both cases. The + version-independent way to ask "how many endpoints exist" is the + resolved OpenAPI schema, which has to fully expand routing to build + itself regardless of the internal representation. + """ import backend.app as app_module - assert len(app_module.app.routes) > 20 + paths = app_module.app.openapi()["paths"] + assert len(paths) > 20 class TestUnaffectedEndpointsStillWork: