update plan

- Add app/api/routes/agents.py with 11 endpoints:
  GET/POST/PUT/DELETE /agents/local (local directory agent configs)
  GET/POST/PUT/DELETE /agents/cloud (cloud connector agent configs)
  GET /agents/catalog (hardcoded agent type catalog)
  GET /agents/runs (paginated run logs with agent_id/page/limit filters)
  POST /agents/{id}/run (manual trigger stub, dispatch wired in step 3.4)
- Tier-gate creation via combined local+cloud batch_active limit
- Ownership checks on all mutations (404 on mismatch)
- Cascade delete of run logs via SQLAlchemy relationship
- Register agents router in app/main.py
- Fix missing import json in app/agents/task_agent.py

AI_REFACTOR_PLAN.md

@@ -240,4 +240,266 @@ Tools must use **camelCase** field names (Drizzle maps them to snake_case intern
 - **Step 2.1 is the point of no return** — after removing LangChain, there's no local AI fallback.
 - **Phase B (backend changes) must land before Phase 1.3–1.5** — Electron needs the bidirectional WS to talk to.
 - **Phase 3 and Phase 4 are independent** — can be parallelized after Phase 2.
+
+---
+
+## Phase 3 — Agent System: Config, Orchestration & Cloud Connectors
+
+> **Objective:** Backend manages all agent configuration, scheduling, orchestration, and cloud data fetching. Two agent types: **Local Directory Agent** (backend triggers Electron to read files, then AI analyzes) and **Cloud Connector Agent** (backend fetches Gmail/Teams data directly, AI analyzes, pushes results to Electron via WS tool_call). All extracted items use existing WS tool infrastructure to insert into Electron's local DB with `is_ai_suggested=True`.
+>
+> **Electron Phase 3 plan:** `../adiuva/AI_REFACTOR_PLAN.md` Phase 3 section.
+
+### Architecture
+
+```
+Local Agent:
+  Scheduler/manual trigger ──► check device online ──► WS agent_run → Electron
+    ──► Electron reads files ──► WS agent_data → Backend
+    ──► Backend AI (prompt_template + file content) ──► WS tool_call(insert) → Electron
+    ──► Electron persists with isAiSuggested=1
+
+Cloud Agent:
+  Scheduler/manual trigger ──► Backend fetches Gmail/Teams (OAuth) ──► Backend AI analyzes
+    ──► check device online ──► WS tool_call(insert) → Electron ──► Electron persists
+```
+
+**New WS frame types:**
+
+| Direction | `type` | Payload |
+|---|---|---|
+| Server → Client | `agent_run` | `{ run_id, agent_id, config: { paths, file_extensions, prompt_template, data_types } }` |
+| Client → Server | `agent_data` | `{ run_id, files: [{ path, name, content, metadata }] }` |
+| Client → Server | `agent_complete` | `{ run_id, files_read, errors }` |
+| Client → Server | `device_hello` | `{ device_id, agent_ids }` |
+
+### Step 3.1 — Agent config tables
+- [x] Add to `app/models.py`:
+  - **`LocalAgentConfig`**:
+    - `id` UUID PK
+    - `user_id` FK → users
+    - `device_id` str — identifies which Electron install this config belongs to
+    - `name` str
+    - `directory_paths` JSON — list of absolute paths on the device
+    - `data_types` JSON — which tables to extract to: `["tasks", "notes", "checkpoints", "projects"]`
+    - `prompt_template` text — user-configured via Chatbot Journey
+    - `file_extensions` JSON — e.g. `[".eml", ".txt", ".pdf", ".md"]`
+    - `schedule_cron` str — e.g. `"0 */6 * * *"` (every 6h)
+    - `enabled` bool (default True)
+    - `last_run_at` datetime nullable
+    - `created_at`, `updated_at` timestamps
+  - **`CloudAgentConfig`**:
+    - `id` UUID PK
+    - `user_id` FK → users
+    - `provider` str — enum: `gmail`, `teams`, `outlook`
+    - `name` str
+    - `data_types` JSON — same format as local
+    - `prompt_template` text
+    - `oauth_token_encrypted` text — Fernet-encrypted OAuth2 credentials
+    - `schedule_cron` str
+    - `enabled` bool (default True)
+    - `last_run_at` datetime nullable
+    - `filter_config` JSON — provider-specific: `{ labels: [], date_range: {from, to}, senders: [] }`
+    - `created_at`, `updated_at` timestamps
+  - **`AgentRunLog`**:
+    - `id` UUID PK
+    - `agent_id` str — references LocalAgentConfig.id or CloudAgentConfig.id
+    - `agent_type` str — `local` or `cloud`
+    - `user_id` FK → users
+    - `status` str — `running`, `success`, `error`, `partial`
+    - `items_processed` int (default 0)
+    - `items_created` int (default 0)
+    - `errors` JSON — list of error strings
+    - `started_at` datetime
+    - `completed_at` datetime nullable
+- [x] Add Pydantic schemas to `app/schemas.py`:
+  - `LocalAgentConfigCreate`, `LocalAgentConfigUpdate`, `LocalAgentConfigResponse`
+  - `CloudAgentConfigCreate`, `CloudAgentConfigUpdate`, `CloudAgentConfigResponse`
+  - `AgentRunLogResponse`
+  - `AgentCatalogItem` — `{ type, name, description, config_schema }`
+  - `WsAgentRun`, `WsAgentData`, `WsAgentComplete`, `WsDeviceHello`
+- [x] Generate Alembic migration
+- **Files:** `app/models.py`, `app/schemas.py`, `alembic/versions/`
+- **Outcome:** Agent config and run tracking tables in PostgreSQL.
+
+### Step 3.2 — Agent CRUD API routes
+- [x] Create `app/api/routes/agents.py`:
+  - `GET /api/v1/agents/catalog` — returns hardcoded agent type catalog:
+    - `local_directory`: "Watches local directories, extracts data from files using AI"
+    - `gmail`: "Scans Gmail inbox, extracts tasks/notes from emails"
+    - `teams`: "Monitors Teams messages, extracts action items"
+    - `outlook`: "Scans Outlook inbox, extracts tasks/notes"
+  - `GET /api/v1/agents/local` — list user's local agent configs
+  - `POST /api/v1/agents/local` — create local agent config
+    - Body: `{ name, device_id, directory_paths, data_types, prompt_template, file_extensions, schedule_cron }`
+    - Tier check: count enabled agents ≤ `batch_active` limit
+  - `PUT /api/v1/agents/local/{id}` — update config (ownership check)
+  - `DELETE /api/v1/agents/local/{id}` — delete config + associated run logs
+  - `GET /api/v1/agents/cloud` — list user's cloud agent configs
+  - `POST /api/v1/agents/cloud` — create cloud connector config
+    - Body: `{ provider, name, data_types, prompt_template, oauth_token_encrypted, schedule_cron, filter_config }`
+    - Tier check: same `batch_active` limit (local + cloud count together)
+  - `PUT /api/v1/agents/cloud/{id}` — update config
+  - `DELETE /api/v1/agents/cloud/{id}` — delete config + run logs
+  - `GET /api/v1/agents/runs` — query params: `agent_id`, `page`, `limit` → paginated run logs
+  - `POST /api/v1/agents/{id}/run` — manual trigger (dispatches to agent runner)
+  - All routes require JWT auth; ownership enforced on all mutations
+- [x] Register router in `app/main.py`
+- **Files:** `app/api/routes/agents.py`, `app/main.py`
+- **Outcome:** Full CRUD for agent configs with tier-gated creation limits.
+
+### Step 3.3 — Device WS endpoint
+- [x] Create `app/api/routes/device_ws.py`:
+  - `WebSocket /api/v1/ws/device?token=<jwt>` — persistent connection from Electron
+  - On connect:
+    - Authenticate JWT
+    - Receive `device_hello` frame → extract `device_id`, `agent_ids`
+    - Store connection in `DeviceConnectionManager` (in-memory dict: `user_id → { ws, device_id }`)
+    - Check for overdue agent runs → trigger them immediately
+  - Message loop:
+    - `agent_data` → route to active agent run handler
+    - `agent_complete` → finalize agent run
+    - `tool_result` → route to pending tool call (same pattern as chat WS)
+    - `pong` → heartbeat ack
+  - On disconnect:
+    - Remove from `DeviceConnectionManager`
+    - Mark any in-progress agent runs as `error` with "device disconnected"
+  - Heartbeat: send `ping` every 30s, disconnect if no `pong` within 10s
+- [x] Create `app/core/device_manager.py`:
+  - `DeviceConnectionManager` (singleton):
+    - `register(user_id, device_id, ws)` — stores active connection
+    - `unregister(user_id)` — removes connection
+    - `get_ws(user_id) -> WebSocket | None` — returns active WS if device is online
+    - `is_online(user_id, device_id=None) -> bool` — optionally checks specific device
+    - `send_frame(user_id, frame: dict)` — sends JSON frame to device
+- **Files:** `app/api/routes/device_ws.py`, `app/core/device_manager.py`, `app/main.py`
+- **Outcome:** Backend maintains persistent WS connections to Electron devices for agent triggers.
+
+### Step 3.4 — Agent run orchestrator
+- [x] Create `app/core/agent_runner.py`:
+  - `async run_local_agent(user_id, config: LocalAgentConfig, device_mgr: DeviceConnectionManager)`:
+    1. Check device is online with matching `device_id` → abort if offline
+    2. Create `AgentRunLog` with `status=running`
+    3. Send `WsAgentRun` frame to Electron with config (paths, extensions, prompt)
+    4. Await `WsAgentData` frames — collect file contents
+    5. Await `WsAgentComplete` frame — Electron signals done reading
+    6. For each file: call LLM with `prompt_template` + file content → extract structured items
+    7. For each extracted item: send `WsToolCall(insert, table, data)` to Electron → await `WsToolResult`
+       - All inserts include `is_ai_suggested=True, is_approved=False`
+    8. Update `AgentRunLog`: `status=success`, `items_processed`, `items_created`
+  - `async run_cloud_agent(user_id, config: CloudAgentConfig, device_mgr: DeviceConnectionManager)`:
+    1. Check device is online → abort if offline (results must push to Electron)
+    2. Create `AgentRunLog` with `status=running`
+    3. Decrypt OAuth credentials from `config.oauth_token_encrypted`
+    4. Fetch data from cloud provider (Step 3.6):
+       - Gmail: `google-api-python-client` + `filter_config` label/date filters
+       - Teams: `msgraph-sdk` + channel/date filters
+       - Outlook: `msgraph-sdk` + folder/date filters
+    5. For each item: call LLM with `prompt_template` + email/message content → extract structured items
+    6. For each extracted item: send `WsToolCall(insert)` to Electron → await `WsToolResult`
+    7. Update `AgentRunLog`
+  - `async trigger_pending_runs(user_id, device_id, device_mgr)`:
+    - Called when Electron connects (after `device_hello`)
+    - Queries all enabled agent configs where `last_run_at + schedule_interval < now()`
+    - For local agents: only triggers if `config.device_id == device_id`
+    - For cloud agents: triggers regardless of device (any connected device can receive results)
+    - Executes runs sequentially (one at a time to avoid overwhelming the WS)
+  - Error handling: on any failure, update `AgentRunLog` with `status=error` + error details
+- [x] Wire `POST /agents/{id}/run` endpoint to dispatch background task via `asyncio.create_task()`
+- [x] Replace `_trigger_pending_runs_stub` in `device_ws.py` with real `trigger_pending_runs` call
+- [x] Add `croniter>=3.0.0` to `requirements.txt`
+- [x] 23 unit + integration tests covering all code paths
+- **Files:** `app/core/agent_runner.py`, `app/api/routes/agents.py`, `app/api/routes/device_ws.py`, `requirements.txt`, `tests/test_agent_runner.py`
+- **Outcome:** Backend drives all agent execution — both local (via WS file request) and cloud (direct API calls — stub until Step 3.6).
+
+### Step 3.5 — Chatbot Journey endpoint
+- [ ] Create `app/api/routes/agent_setup.py`:
+  - `POST /api/v1/agents/journey/start`:
+    - Body: `{ agent_type: "local"|"cloud", data_types: ["tasks", "notes", ...] }`
+    - Creates a journey session (in-memory or Redis-backed)
+    - Returns first AI message: contextual question based on agent type
+      - Local: "What kind of files are in the directories you want to monitor? (emails, documents, logs, etc.)"
+      - Cloud: "What kind of emails/messages should I look for? (client communications, invoices, meeting notes, etc.)"
+    - Response: `{ session_id, message, done: false }`
+  - `POST /api/v1/agents/journey/message`:
+    - Body: `{ session_id, message }`
+    - AI processes user's answer, asks follow-up questions (max 5 turns)
+    - System prompt: "You are configuring a data extraction agent for a freelancer. Ask about file format, what data to extract (tasks, notes, checkpoints), naming conventions, priority rules, and any special mapping. After 3-5 questions, generate a detailed prompt_template."
+    - When AI determines enough context: `{ session_id, message: "Here's your configuration...", done: true, prompt_template: "..." }`
+    - The `prompt_template` is a structured instruction for the extraction LLM (e.g. "Extract tasks from email. Subject becomes task title. If body contains 'urgent' or 'ASAP', set priority to 'high'. Extract due dates if mentioned.")
+- **Files:** `app/api/routes/agent_setup.py`, `app/main.py`
+- **Outcome:** Users configure AI prompts through guided conversation, not manual text editing.
+
+### Step 3.6 — Cloud provider integrations
+- [ ] Create `app/integrations/gmail.py`:
+  - `GmailClient`:
+    - `__init__(oauth_token)` — initializes Google API client
+    - `async fetch_messages(filter_config, since: datetime) -> list[EmailMessage]`
+    - `EmailMessage`: `{ id, subject, sender, body_text, date, labels }`
+    - Handles token refresh via Google OAuth2 refresh flow
+    - Respects `filter_config.labels`, `filter_config.date_range`, `filter_config.senders`
+- [ ] Create `app/integrations/ms_graph.py`:
+  - `MSGraphClient`:
+    - `__init__(oauth_token)` — initializes MS Graph client
+    - `async fetch_emails(filter_config, since: datetime) -> list[EmailMessage]` (Outlook)
+    - `async fetch_messages(filter_config, since: datetime) -> list[ChatMessage]` (Teams)
+    - `ChatMessage`: `{ id, content, sender, channel, date }`
+    - Handles token refresh via MSAL
+- [ ] Create `app/integrations/__init__.py` — factory: `get_provider(provider_name) -> GmailClient | MSGraphClient`
+- **Dependencies:** `google-api-python-client`, `google-auth-oauthlib`, `msgraph-sdk`, `msal`
+- **Files:** `app/integrations/gmail.py`, `app/integrations/ms_graph.py`, `app/integrations/__init__.py`
+- **Outcome:** Backend can fetch emails/messages from Gmail, Outlook, and Teams.
+
+### Step 3.7 — Agent scheduler
+- [ ] Create `app/core/agent_scheduler.py`:
+  - Uses `APScheduler` (or simple asyncio loop) to check agent schedules
+  - Every 60s: query enabled agents where `last_run_at + cron_interval < now()`
+  - For each due agent:
+    - Check if user's device is online via `DeviceConnectionManager`
+    - If online: dispatch to `agent_runner`
+    - If offline: skip (will trigger on next `device_hello`)
+  - Locks: use PostgreSQL advisory locks to prevent duplicate runs in multi-instance deployments
+- [ ] Integrate with FastAPI lifespan (start scheduler on app startup, shutdown gracefully)
+- **Dependencies:** `apscheduler>=4.0`
+- **Files:** `app/core/agent_scheduler.py`, `app/main.py`
+- **Outcome:** Agents run automatically on their configured schedules.
+
+### Step 3.8 — OAuth flow endpoints
+- [ ] Create `app/api/routes/oauth.py`:
+  - `GET /api/v1/oauth/{provider}/authorize` — returns OAuth authorization URL
+    - Gmail: Google OAuth2 with `gmail.readonly` scope
+    - Outlook/Teams: MS identity platform with `Mail.Read`, `ChannelMessage.Read.All` scopes
+  - `GET /api/v1/oauth/{provider}/callback` — handles OAuth redirect
+    - Exchanges auth code for access + refresh tokens
+    - Encrypts tokens with Fernet (server-side key from settings)
+    - Returns encrypted token blob for storage in `CloudAgentConfig.oauth_token_encrypted`
+  - `POST /api/v1/oauth/{provider}/refresh` — refresh expired OAuth token
+- **Files:** `app/api/routes/oauth.py`, `app/main.py`
+- **Outcome:** Users can connect Gmail/Teams/Outlook accounts securely.
+
+---
+
+### Phase 3 — Verification
+
+| # | Scenario | Expected |
+|---|---|---|
+| 1 | **Agent CRUD** | Create/read/update/delete local and cloud configs; tier limits enforced (free=2, pro=10) |
+| 2 | **WS device connect** | Electron connects → `device_hello` → backend stores connection → triggers overdue runs |
+| 3 | **Local agent run** | Backend sends `agent_run` → Electron reads files → `agent_data` → backend AI extracts → `tool_call(insert)` → Electron persists with `isAiSuggested=1` |
+| 4 | **Cloud agent run** | Backend fetches Gmail → AI extracts tasks → `tool_call(insert)` → Electron persists |
+| 5 | **Device binding** | Local agent config with `device_id=A` only triggers when device A is connected |
+| 6 | **Chatbot Journey** | Start journey → 3-5 Q&A turns → produces valid `prompt_template` |
+| 7 | **Schedule** | Agent with `schedule_cron="0 */6 * * *"` runs every 6h when device is online |
+| 8 | **Offline resilience** | Device offline → runs skipped → device reconnects → overdue runs trigger immediately |
+| 9 | **OAuth flow** | Gmail authorize → callback → token encrypted → stored in config → fetch emails works |
+
+### Phase 3 — New Dependencies
+
+| Package | Purpose |
+|---|---|
+| `google-api-python-client` | Gmail API access |
+| `google-auth-oauthlib` | Gmail OAuth2 flow |
+| `msgraph-sdk` | Outlook + Teams API access |
+| `msal` | MS identity platform auth |
+| `apscheduler>=4.0` | Agent scheduling |
+| `cryptography` (Fernet) | OAuth token encryption at rest |
 - **One step at a time.** Mark `[x]` and commit with `step N.N complete: <outcome>`.

BACKEND_PLAN.md

@@ -500,6 +500,22 @@ adiuva-api/
 | GET | `/api/v1/billing/subscription` | JWT | — | Subscription info |
 | DELETE | `/api/v1/billing/subscription` | JWT | — | `{ok: true}` |
 | GET | `/api/v1/health` | No | — | `{status, version}` |
+| GET | `/api/v1/agents/catalog` | JWT | — | `AgentCatalogItem[]` |
+| GET | `/api/v1/agents/local` | JWT | — | `LocalAgentConfigResponse[]` |
+| POST | `/api/v1/agents/local` | JWT | `LocalAgentConfigCreate` | `LocalAgentConfigResponse` |
+| PUT | `/api/v1/agents/local/{id}` | JWT | `LocalAgentConfigUpdate` | `LocalAgentConfigResponse` |
+| DELETE | `/api/v1/agents/local/{id}` | JWT | — | `{ok: true}` |
+| GET | `/api/v1/agents/cloud` | JWT | — | `CloudAgentConfigResponse[]` |
+| POST | `/api/v1/agents/cloud` | JWT | `CloudAgentConfigCreate` | `CloudAgentConfigResponse` |
+| PUT | `/api/v1/agents/cloud/{id}` | JWT | `CloudAgentConfigUpdate` | `CloudAgentConfigResponse` |
+| DELETE | `/api/v1/agents/cloud/{id}` | JWT | — | `{ok: true}` |
+| GET | `/api/v1/agents/runs` | JWT | `?agent_id&page&limit` | `AgentRunLogResponse[]` |
+| POST | `/api/v1/agents/{id}/run` | JWT | — | `{ok: true, run_id}` |
+| POST | `/api/v1/agents/journey/start` | JWT | `{agent_type, data_types}` | `{session_id, message, done}` |
+| POST | `/api/v1/agents/journey/message` | JWT | `{session_id, message}` | `{session_id, message, done, prompt_template?}` |
+| GET | `/api/v1/oauth/{provider}/authorize` | JWT | — | `{authorization_url}` |
+| GET | `/api/v1/oauth/{provider}/callback` | — | OAuth code | `{encrypted_token}` |
+| WS | `/api/v1/ws/device` | JWT | `device_hello` (first frame) | Agent trigger + tool_call frames |
 
 ---
 
@@ -515,11 +531,34 @@ adiuva-api/
 | Vector store | Pinecone or Qdrant (configurable) |
 | Database | PostgreSQL + SQLAlchemy + Alembic |
 | Rate limiting | slowapi |
+| Cloud integrations | google-api-python-client, msgraph-sdk, msal |
+| Agent scheduling | APScheduler |
 | Testing | pytest + pytest-asyncio + httpx + moto (S3 mock) |
 | Deployment | Docker → fly.io / Railway / AWS ECS |
 
 ---
 
+## Phase 3 — New Files
+
+| File | Purpose |
+|---|---|
+| `app/models.py` | Add `LocalAgentConfig`, `CloudAgentConfig`, `AgentRunLog` models |
+| `app/schemas.py` | Add agent config schemas + WS agent frame types |
+| `app/api/routes/agents.py` | Agent CRUD endpoints (catalog, local, cloud, runs, manual trigger) |
+| `app/api/routes/agent_setup.py` | Chatbot Journey endpoints (start + message) |
+| `app/api/routes/device_ws.py` | Persistent device WS endpoint (`/api/v1/ws/device`) |
+| `app/api/routes/oauth.py` | OAuth authorize/callback for Gmail, Teams, Outlook |
+| `app/core/agent_runner.py` | Agent run orchestration — local (WS file request) + cloud (API fetch) |
+| `app/core/device_manager.py` | `DeviceConnectionManager` — tracks active Electron WS connections |
+| `app/core/agent_scheduler.py` | Periodic scheduler for agent cron triggers |
+| `app/integrations/gmail.py` | Gmail API client (fetch messages with filters) |
+| `app/integrations/ms_graph.py` | MS Graph client for Outlook emails + Teams messages |
+| `app/integrations/__init__.py` | Provider factory |
+
+> **Full Phase 3 step-by-step plan:** See `AI_REFACTOR_PLAN.md` Phase 3 section.
+
+---
+
 ## Development Rules
 
 1. **NEVER persist user data in plaintext.** The DB stores only auth, billing, storage metadata, and marketplace data. User context arrives in requests and is discarded. Cloud blobs are E2E encrypted client-side — backend only stores opaque bytes.

alembic/versions/003_agent_tables.py

@@ -0,0 +1,127 @@
+"""Add agent config and run log tables: local_agent_configs, cloud_agent_configs, agent_run_logs.
+
+Revision ID: 003
+Revises: 002
+Create Date: 2026-03-05
+"""
+
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "003"
+down_revision: Union[str, None] = "002"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ── Enum types — idempotent creation ──────────────────────────────────
+    op.execute("""
+        DO $$ BEGIN
+            CREATE TYPE agent_type AS ENUM ('local', 'cloud');
+        EXCEPTION WHEN duplicate_object THEN NULL;
+        END $$;
+    """)
+    op.execute("""
+        DO $$ BEGIN
+            CREATE TYPE agent_run_status AS ENUM ('running', 'success', 'error', 'partial');
+        EXCEPTION WHEN duplicate_object THEN NULL;
+        END $$;
+    """)
+    op.execute("""
+        DO $$ BEGIN
+            CREATE TYPE cloud_provider AS ENUM ('gmail', 'teams', 'outlook');
+        EXCEPTION WHEN duplicate_object THEN NULL;
+        END $$;
+    """)
+
+    # ── local_agent_configs ───────────────────────────────────────────────
+    op.create_table(
+        "local_agent_configs",
+        sa.Column("id", postgresql.UUID(as_uuid=False), nullable=False),
+        sa.Column("user_id", postgresql.UUID(as_uuid=False), nullable=False),
+        sa.Column("device_id", sa.String(255), nullable=False),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("directory_paths", sa.JSON, nullable=False, server_default="[]"),
+        sa.Column("data_types", sa.JSON, nullable=False, server_default="[]"),
+        sa.Column("prompt_template", sa.Text, nullable=False, server_default=""),
+        sa.Column("file_extensions", sa.JSON, nullable=False, server_default="[]"),
+        sa.Column("schedule_cron", sa.String(100), nullable=False, server_default="0 */6 * * *"),
+        sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.true()),
+        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
+    )
+    op.create_index("ix_local_agent_configs_user_id", "local_agent_configs", ["user_id"])
+
+    # ── cloud_agent_configs ───────────────────────────────────────────────
+    op.create_table(
+        "cloud_agent_configs",
+        sa.Column("id", postgresql.UUID(as_uuid=False), nullable=False),
+        sa.Column("user_id", postgresql.UUID(as_uuid=False), nullable=False),
+        sa.Column(
+            "provider",
+            postgresql.ENUM("gmail", "teams", "outlook", name="cloud_provider", create_type=False),
+            nullable=False,
+        ),
+        sa.Column("name", sa.String(255), nullable=False),
+        sa.Column("data_types", sa.JSON, nullable=False, server_default="[]"),
+        sa.Column("prompt_template", sa.Text, nullable=False, server_default=""),
+        sa.Column("oauth_token_encrypted", sa.Text, nullable=True),
+        sa.Column("filter_config", sa.JSON, nullable=True),
+        sa.Column("schedule_cron", sa.String(100), nullable=False, server_default="0 */6 * * *"),
+        sa.Column("enabled", sa.Boolean, nullable=False, server_default=sa.true()),
+        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
+    )
+    op.create_index("ix_cloud_agent_configs_user_id", "cloud_agent_configs", ["user_id"])
+
+    # ── agent_run_logs ─────────────────────────────────────────────────────
+    op.create_table(
+        "agent_run_logs",
+        sa.Column("id", postgresql.UUID(as_uuid=False), nullable=False),
+        # Plain string — not a FK because it references either local_agent_configs or
+        # cloud_agent_configs depending on agent_type.
+        sa.Column("agent_id", sa.String(255), nullable=False),
+        sa.Column(
+            "agent_type",
+            postgresql.ENUM("local", "cloud", name="agent_type", create_type=False),
+            nullable=False,
+        ),
+        sa.Column("user_id", postgresql.UUID(as_uuid=False), nullable=False),
+        sa.Column(
+            "status",
+            postgresql.ENUM("running", "success", "error", "partial", name="agent_run_status", create_type=False),
+            nullable=False,
+            server_default="running",
+        ),
+        sa.Column("items_processed", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("items_created", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("errors", sa.JSON, nullable=True),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
+        sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"], ondelete="CASCADE"),
+    )
+    op.create_index("ix_agent_run_logs_user_id", "agent_run_logs", ["user_id"])
+    op.create_index("ix_agent_run_logs_agent_id", "agent_run_logs", ["agent_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("agent_run_logs")
+    op.drop_table("cloud_agent_configs")
+    op.drop_table("local_agent_configs")
+
+    op.execute("DROP TYPE IF EXISTS cloud_provider;")
+    op.execute("DROP TYPE IF EXISTS agent_run_status;")
+    op.execute("DROP TYPE IF EXISTS agent_type;")

app/agents/task_agent.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from datetime import datetime, timezone
 from typing import Any
 

app/api/routes/agents.py

@@ -0,0 +1,452 @@
+"""Agent CRUD routes: local directory agents and cloud connector agents.
+
+Endpoints:
+  GET    /agents/catalog            — hardcoded agent type catalog
+  GET    /agents/local              — list user's local agent configs
+  POST   /agents/local              — create local agent (tier-gated)
+  PUT    /agents/local/{agent_id}   — partial update (ownership check)
+  DELETE /agents/local/{agent_id}   — delete + cascade run logs
+  GET    /agents/cloud              — list user's cloud agent configs
+  POST   /agents/cloud              — create cloud agent (tier-gated)
+  PUT    /agents/cloud/{agent_id}   — partial update (ownership check)
+  DELETE /agents/cloud/{agent_id}   — delete + cascade run logs
+  GET    /agents/runs               — paginated run logs (agent_id, page, limit)
+  POST   /agents/{agent_id}/run     — manual trigger stub (dispatch in Step 3.4)
+"""
+
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from pydantic import BaseModel
+from sqlalchemy import func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import get_current_user
+from app.billing.tier_manager import FEATURES
+from app.core.agent_runner import run_cloud_agent, run_local_agent
+from app.core.device_manager import device_manager
+from app.db import get_session
+from app.models import AgentRunLog, CloudAgentConfig, LocalAgentConfig
+from app.schemas import (
+    AgentCatalogItem,
+    AgentRunLogResponse,
+    CloudAgentConfigCreate,
+    CloudAgentConfigResponse,
+    CloudAgentConfigUpdate,
+    LocalAgentConfigCreate,
+    LocalAgentConfigResponse,
+    LocalAgentConfigUpdate,
+    UserProfile,
+)
+
+router = APIRouter(prefix="/agents", tags=["agents"])
+
+
+# ── Datetime helpers ──────────────────────────────────────────────────
+
+def _dt_ms(dt: datetime) -> int:
+    return int(dt.timestamp() * 1000)
+
+
+def _dt_ms_opt(dt: datetime | None) -> int | None:
+    return int(dt.timestamp() * 1000) if dt else None
+
+
+# ── Model → schema converters ─────────────────────────────────────────
+
+def _to_local_response(a: LocalAgentConfig) -> LocalAgentConfigResponse:
+    return LocalAgentConfigResponse(
+        id=a.id,
+        name=a.name,
+        device_id=a.device_id,
+        directory_paths=a.directory_paths,
+        data_types=a.data_types,
+        prompt_template=a.prompt_template,
+        file_extensions=a.file_extensions,
+        schedule_cron=a.schedule_cron,
+        enabled=a.enabled,
+        last_run_at=_dt_ms_opt(a.last_run_at),
+        created_at=_dt_ms(a.created_at),
+        updated_at=_dt_ms(a.updated_at),
+    )
+
+
+def _to_cloud_response(a: CloudAgentConfig) -> CloudAgentConfigResponse:
+    return CloudAgentConfigResponse(
+        id=a.id,
+        provider=a.provider,  # type: ignore[arg-type]
+        name=a.name,
+        data_types=a.data_types,
+        prompt_template=a.prompt_template,
+        schedule_cron=a.schedule_cron,
+        filter_config=a.filter_config,
+        enabled=a.enabled,
+        last_run_at=_dt_ms_opt(a.last_run_at),
+        created_at=_dt_ms(a.created_at),
+        updated_at=_dt_ms(a.updated_at),
+    )
+
+
+def _to_run_log_response(log: AgentRunLog) -> AgentRunLogResponse:
+    return AgentRunLogResponse(
+        id=log.id,
+        agent_id=log.agent_id,
+        agent_type=log.agent_type,  # type: ignore[arg-type]
+        status=log.status,  # type: ignore[arg-type]
+        items_processed=log.items_processed,
+        items_created=log.items_created,
+        errors=log.errors or [],
+        started_at=_dt_ms(log.started_at),
+        completed_at=_dt_ms_opt(log.completed_at),
+    )
+
+
+# ── Ownership-checked lookups ─────────────────────────────────────────
+
+async def _get_local_agent_for_user(
+    agent_id: str, user_id: str, db: AsyncSession
+) -> LocalAgentConfig:
+    result = await db.execute(
+        select(LocalAgentConfig).where(
+            LocalAgentConfig.id == agent_id,
+            LocalAgentConfig.user_id == user_id,
+        )
+    )
+    record = result.scalar_one_or_none()
+    if record is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Agent not found")
+    return record
+
+
+async def _get_cloud_agent_for_user(
+    agent_id: str, user_id: str, db: AsyncSession
+) -> CloudAgentConfig:
+    result = await db.execute(
+        select(CloudAgentConfig).where(
+            CloudAgentConfig.id == agent_id,
+            CloudAgentConfig.user_id == user_id,
+        )
+    )
+    record = result.scalar_one_or_none()
+    if record is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Agent not found")
+    return record
+
+
+# ── Tier limit helper ─────────────────────────────────────────────────
+
+async def _count_enabled_agents(user_id: str, db: AsyncSession) -> int:
+    """Return combined enabled local + cloud agent count for the user."""
+    local_count = (
+        await db.execute(
+            select(func.count(LocalAgentConfig.id)).where(
+                LocalAgentConfig.user_id == user_id,
+                LocalAgentConfig.enabled == True,  # noqa: E712
+            )
+        )
+    ).scalar_one()
+    cloud_count = (
+        await db.execute(
+            select(func.count(CloudAgentConfig.id)).where(
+                CloudAgentConfig.user_id == user_id,
+                CloudAgentConfig.enabled == True,  # noqa: E712
+            )
+        )
+    ).scalar_one()
+    return local_count + cloud_count
+
+
+def _enforce_agent_limit(tier: str, current_count: int) -> None:
+    limit: int = FEATURES.get(tier, FEATURES["free"])["batch_active"]
+    if limit != -1 and current_count >= limit:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail=f"Agent limit ({limit}) reached for your tier. Upgrade to create more.",
+        )
+
+
+# ── Local page schema (used by runs endpoint) ─────────────────────────
+
+class _RunsPage(BaseModel):
+    total: int
+    page: int
+    limit: int
+    items: list[AgentRunLogResponse]
+
+
+# ── Catalog ───────────────────────────────────────────────────────────
+
+@router.get("/catalog", response_model=list[AgentCatalogItem])
+async def get_agent_catalog(
+    current_user: UserProfile = Depends(get_current_user),
+) -> list[AgentCatalogItem]:
+    """Return the static list of available agent types and their descriptions."""
+    return [
+        AgentCatalogItem(
+            type="local_directory",
+            name="Local Directory Monitor",
+            description="Watches local directories, extracts data from files using AI",
+        ),
+        AgentCatalogItem(
+            type="gmail",
+            name="Gmail Connector",
+            description="Scans Gmail inbox, extracts tasks/notes from emails",
+        ),
+        AgentCatalogItem(
+            type="teams",
+            name="Microsoft Teams Connector",
+            description="Monitors Teams messages, extracts action items",
+        ),
+        AgentCatalogItem(
+            type="outlook",
+            name="Outlook Connector",
+            description="Scans Outlook inbox, extracts tasks/notes",
+        ),
+    ]
+
+
+# ── Local agent CRUD ──────────────────────────────────────────────────
+
+@router.get("/local", response_model=list[LocalAgentConfigResponse])
+async def list_local_agents(
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> list[LocalAgentConfigResponse]:
+    """List all local directory agent configs owned by the authenticated user."""
+    result = await db.execute(
+        select(LocalAgentConfig).where(LocalAgentConfig.user_id == current_user.id)
+    )
+    return [_to_local_response(a) for a in result.scalars().all()]
+
+
+@router.post("/local", response_model=LocalAgentConfigResponse, status_code=status.HTTP_201_CREATED)
+async def create_local_agent(
+    body: LocalAgentConfigCreate,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> LocalAgentConfigResponse:
+    """Create a new local directory agent config.
+
+    The combined count of enabled local and cloud agents for the user is
+    checked against the ``batch_active`` limit for their billing tier.
+    """
+    _enforce_agent_limit(current_user.tier, await _count_enabled_agents(current_user.id, db))
+    agent = LocalAgentConfig(
+        user_id=current_user.id,
+        name=body.name,
+        device_id=body.device_id,
+        directory_paths=body.directory_paths,
+        data_types=body.data_types,
+        prompt_template=body.prompt_template,
+        file_extensions=body.file_extensions,
+        schedule_cron=body.schedule_cron,
+    )
+    db.add(agent)
+    await db.commit()
+    await db.refresh(agent)
+    return _to_local_response(agent)
+
+
+@router.put("/local/{agent_id}", response_model=LocalAgentConfigResponse)
+async def update_local_agent(
+    agent_id: str,
+    body: LocalAgentConfigUpdate,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> LocalAgentConfigResponse:
+    """Partially update a local agent config. Only provided fields are changed."""
+    agent = await _get_local_agent_for_user(agent_id, current_user.id, db)
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(agent, field, value)
+    await db.commit()
+    await db.refresh(agent)
+    return _to_local_response(agent)
+
+
+@router.delete("/local/{agent_id}", response_model=dict)
+async def delete_local_agent(
+    agent_id: str,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> dict[str, bool]:
+    """Delete a local agent config. Associated run logs are cascade-deleted."""
+    agent = await _get_local_agent_for_user(agent_id, current_user.id, db)
+    await db.delete(agent)
+    await db.commit()
+    return {"ok": True}
+
+
+# ── Cloud agent CRUD ──────────────────────────────────────────────────
+
+@router.get("/cloud", response_model=list[CloudAgentConfigResponse])
+async def list_cloud_agents(
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> list[CloudAgentConfigResponse]:
+    """List all cloud connector agent configs owned by the authenticated user."""
+    result = await db.execute(
+        select(CloudAgentConfig).where(CloudAgentConfig.user_id == current_user.id)
+    )
+    return [_to_cloud_response(a) for a in result.scalars().all()]
+
+
+@router.post("/cloud", response_model=CloudAgentConfigResponse, status_code=status.HTTP_201_CREATED)
+async def create_cloud_agent(
+    body: CloudAgentConfigCreate,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> CloudAgentConfigResponse:
+    """Create a new cloud connector agent config.
+
+    The combined count of enabled local and cloud agents for the user is
+    checked against the ``batch_active`` limit for their billing tier.
+    """
+    _enforce_agent_limit(current_user.tier, await _count_enabled_agents(current_user.id, db))
+    agent = CloudAgentConfig(
+        user_id=current_user.id,
+        provider=body.provider,
+        name=body.name,
+        data_types=body.data_types,
+        prompt_template=body.prompt_template,
+        oauth_token_encrypted=body.oauth_token_encrypted,
+        schedule_cron=body.schedule_cron,
+        filter_config=body.filter_config,
+    )
+    db.add(agent)
+    await db.commit()
+    await db.refresh(agent)
+    return _to_cloud_response(agent)
+
+
+@router.put("/cloud/{agent_id}", response_model=CloudAgentConfigResponse)
+async def update_cloud_agent(
+    agent_id: str,
+    body: CloudAgentConfigUpdate,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> CloudAgentConfigResponse:
+    """Partially update a cloud agent config. Only provided fields are changed."""
+    agent = await _get_cloud_agent_for_user(agent_id, current_user.id, db)
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(agent, field, value)
+    await db.commit()
+    await db.refresh(agent)
+    return _to_cloud_response(agent)
+
+
+@router.delete("/cloud/{agent_id}", response_model=dict)
+async def delete_cloud_agent(
+    agent_id: str,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> dict[str, bool]:
+    """Delete a cloud agent config. Associated run logs are cascade-deleted."""
+    agent = await _get_cloud_agent_for_user(agent_id, current_user.id, db)
+    await db.delete(agent)
+    await db.commit()
+    return {"ok": True}
+
+
+# ── Run logs ──────────────────────────────────────────────────────────
+
+@router.get("/runs", response_model=_RunsPage)
+async def list_run_logs(
+    agent_id: str | None = Query(default=None),
+    page: int = Query(default=1, ge=1),
+    limit: int = Query(default=20, ge=1, le=100),
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> _RunsPage:
+    """Return paginated run logs for the authenticated user.
+
+    Optionally filter by ``agent_id``. Results are ordered from newest to oldest.
+    """
+    base_filter = [AgentRunLog.user_id == current_user.id]
+    if agent_id:
+        base_filter.append(AgentRunLog.agent_id == agent_id)
+
+    total = (
+        await db.execute(select(func.count(AgentRunLog.id)).where(*base_filter))
+    ).scalar_one()
+
+    result = await db.execute(
+        select(AgentRunLog)
+        .where(*base_filter)
+        .order_by(AgentRunLog.started_at.desc())
+        .offset((page - 1) * limit)
+        .limit(limit)
+    )
+    items = [_to_run_log_response(log) for log in result.scalars().all()]
+
+    return _RunsPage(total=total, page=page, limit=limit, items=items)
+
+
+# ── Manual trigger stub ───────────────────────────────────────────────
+
+@router.post("/{agent_id}/run", response_model=AgentRunLogResponse, status_code=status.HTTP_202_ACCEPTED)
+async def trigger_agent_run(
+    agent_id: str,
+    current_user: UserProfile = Depends(get_current_user),
+    db: AsyncSession = Depends(get_session),
+) -> AgentRunLogResponse:
+    """Manually trigger an agent run.
+
+    Looks up the agent config (local or cloud) by ID with ownership check,
+    creates a run log entry with ``status="running"``, and returns it.
+
+    Actual dispatch to the agent runner is wired in Step 3.4 once
+    ``DeviceConnectionManager`` and ``agent_runner`` are available.
+    """
+    # Determine agent type by trying local first, then cloud.
+    # Keep the full config object so we can pass it to the agent runner.
+    local_config: LocalAgentConfig | None = None
+    cloud_config: CloudAgentConfig | None = None
+
+    local_result = await db.execute(
+        select(LocalAgentConfig).where(
+            LocalAgentConfig.id == agent_id,
+            LocalAgentConfig.user_id == current_user.id,
+        )
+    )
+    local_config = local_result.scalar_one_or_none()
+
+    if local_config is not None:
+        agent_type = "local"
+    else:
+        cloud_result = await db.execute(
+            select(CloudAgentConfig).where(
+                CloudAgentConfig.id == agent_id,
+                CloudAgentConfig.user_id == current_user.id,
+            )
+        )
+        cloud_config = cloud_result.scalar_one_or_none()
+        if cloud_config is not None:
+            agent_type = "cloud"
+        else:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Agent not found")
+
+    run_log = AgentRunLog(
+        agent_id=agent_id,
+        agent_type=agent_type,
+        user_id=current_user.id,
+        status="running",
+    )
+    db.add(run_log)
+    await db.commit()
+    await db.refresh(run_log)
+
+    # Dispatch the run as a background task — returns 202 immediately.
+    if agent_type == "local" and local_config is not None:
+        asyncio.create_task(
+            run_local_agent(current_user.id, local_config, run_log, device_manager)
+        )
+    elif agent_type == "cloud" and cloud_config is not None:
+        asyncio.create_task(
+            run_cloud_agent(current_user.id, cloud_config, run_log, device_manager)
+        )
+
+    return _to_run_log_response(run_log)

app/api/routes/device_ws.py

@@ -0,0 +1,221 @@
+"""Device WebSocket endpoint.
+
+Persistent connection from Electron devices to the backend.
+
+  WS  /api/v1/ws/device?token=<jwt>
+
+Auth: JWT passed as ``?token=`` query parameter (Bearer header is not
+available during the WebSocket handshake).
+
+Protocol:
+  1. Client connects → JWT validated → connection accepted.
+  2. Client sends ``device_hello`` frame: ``{ type, device_id, agent_ids }``.
+  3. Backend registers the connection in ``DeviceConnectionManager``.
+  4. Session enters message dispatch loop + heartbeat.
+
+Incoming frame dispatch:
+  - ``tool_result``    → resolves a pending tool-call Future.
+  - ``agent_data``     → enqueued in the per-run agent data queue.
+  - ``agent_complete`` → sends None sentinel to close the queue stream.
+  - ``pong``           → heartbeat acknowledgement (updates last-seen).
+  - unknown types      → logged, ignored.
+
+Outgoing heartbeat: ``{ "type": "ping" }`` every 30 s.
+
+On disconnect:
+  - Unregisters from DeviceConnectionManager.
+  - Marks all in-progress AgentRunLog rows for this user as ``error``
+    with message "device disconnected".
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+from jose import JWTError, jwt
+from sqlalchemy import select, update
+
+from app.config.settings import settings
+from app.core.agent_runner import trigger_pending_runs
+from app.core.device_manager import device_manager
+from app.db import async_session
+from app.models import AgentRunLog
+from app.schemas import WsFrameType
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/ws", tags=["device-ws"])
+
+_HEARTBEAT_INTERVAL = 30  # seconds
+_PONG_TIMEOUT = 10  # seconds — grace window after a ping
+
+
+@router.websocket("/device")
+async def device_ws(websocket: WebSocket) -> None:
+    """Persistent WebSocket endpoint for Electron device connections.
+
+    Authentication is via ``?token=<jwt>`` query parameter.
+    """
+    # ── 1. Authenticate before accepting ─────────────────────────────
+    token = websocket.query_params.get("token", "")
+    try:
+        payload = jwt.decode(
+            token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM]
+        )
+        user_id: str | None = payload.get("sub")
+        if not user_id:
+            raise JWTError("missing sub")
+    except JWTError:
+        await websocket.close(code=1008)  # Policy Violation
+        return
+
+    await websocket.accept()
+
+    # ── 2. Await device_hello frame ───────────────────────────────────
+    try:
+        raw = await asyncio.wait_for(websocket.receive_text(), timeout=15.0)
+    except (asyncio.TimeoutError, WebSocketDisconnect):
+        await websocket.close(code=1008)
+        return
+
+    try:
+        hello = json.loads(raw)
+        if hello.get("type") != WsFrameType.device_hello:
+            raise ValueError("expected device_hello as first frame")
+        device_id: str = hello["device_id"]
+        agent_ids: list[str] = hello.get("agent_ids", [])
+    except (KeyError, ValueError, json.JSONDecodeError) as exc:
+        logger.warning("device_ws: invalid device_hello from user=%s: %s", user_id, exc)
+        await websocket.close(code=1008)
+        return
+
+    # ── 3. Register connection ────────────────────────────────────────
+    device_manager.register(user_id, device_id, websocket)
+    logger.info(
+        "device_ws: connected user=%s device=%s agents=%s",
+        user_id,
+        device_id,
+        agent_ids,
+    )
+
+    # Trigger any overdue agent runs now that the device is connected.
+    asyncio.create_task(trigger_pending_runs(user_id, device_id, device_manager))
+
+    # ── 4. Concurrent message loop + heartbeat ────────────────────────
+    try:
+        await asyncio.gather(
+            _message_loop(websocket, user_id),
+            _heartbeat_loop(websocket),
+        )
+    except WebSocketDisconnect:
+        pass
+    except Exception as exc:
+        logger.warning("device_ws: unhandled exception user=%s: %s", user_id, exc)
+    finally:
+        device_manager.unregister(user_id)
+        logger.info("device_ws: disconnected user=%s device=%s", user_id, device_id)
+        await _mark_runs_disconnected(user_id)
+
+
+# ── Message dispatch loop ─────────────────────────────────────────────
+
+async def _message_loop(websocket: WebSocket, user_id: str) -> None:
+    """Receive frames from Electron and dispatch to the appropriate handler."""
+    async for raw in websocket.iter_text():
+        try:
+            frame: dict = json.loads(raw)
+        except json.JSONDecodeError:
+            logger.warning("device_ws: invalid JSON from user=%s", user_id)
+            continue
+
+        frame_type = frame.get("type")
+
+        if frame_type == WsFrameType.tool_result:
+            call_id = frame.get("id")
+            if call_id:
+                device_manager.resolve_pending_call(user_id, call_id, frame)
+            else:
+                logger.warning(
+                    "device_ws: tool_result missing id from user=%s", user_id
+                )
+
+        elif frame_type == WsFrameType.agent_data:
+            run_id = frame.get("run_id")
+            if run_id:
+                try:
+                    queue = device_manager.get_agent_data_queue(user_id, run_id)
+                    await queue.put(frame)
+                except RuntimeError:
+                    logger.warning(
+                        "device_ws: agent_data for unknown run user=%s run=%s",
+                        user_id,
+                        run_id,
+                    )
+            else:
+                logger.warning(
+                    "device_ws: agent_data missing run_id from user=%s", user_id
+                )
+
+        elif frame_type == WsFrameType.agent_complete:
+            run_id = frame.get("run_id")
+            if run_id:
+                try:
+                    queue = device_manager.get_agent_data_queue(user_id, run_id)
+                    # Sentinel: signals the agent data stream is finished.
+                    await queue.put(None)
+                except RuntimeError:
+                    pass
+            else:
+                logger.warning(
+                    "device_ws: agent_complete missing run_id from user=%s", user_id
+                )
+
+        elif frame_type == "pong":
+            # Heartbeat ack — nothing to do, connection is alive.
+            pass
+
+        else:
+            logger.debug(
+                "device_ws: unknown frame type %r from user=%s", frame_type, user_id
+            )
+
+
+# ── Heartbeat ─────────────────────────────────────────────────────────
+
+async def _heartbeat_loop(websocket: WebSocket) -> None:
+    """Send a ping frame every 30 s to keep the connection alive."""
+    while True:
+        await asyncio.sleep(_HEARTBEAT_INTERVAL)
+        await websocket.send_text(json.dumps({"type": "ping"}))
+
+
+# ── Disconnect cleanup ────────────────────────────────────────────────
+
+async def _mark_runs_disconnected(user_id: str) -> None:
+    """Mark all in-progress AgentRunLog rows as 'error' for this user."""
+    try:
+        async with async_session() as db:
+            await db.execute(
+                update(AgentRunLog)
+                .where(
+                    AgentRunLog.user_id == user_id,
+                    AgentRunLog.status == "running",
+                )
+                .values(
+                    status="error",
+                    errors=["device disconnected"],
+                )
+            )
+            await db.commit()
+    except Exception as exc:
+        logger.error(
+            "device_ws: failed to mark runs as disconnected for user=%s: %s",
+            user_id,
+            exc,
+        )
+
+
+

app/core/agent_runner.py

@@ -0,0 +1,534 @@
+"""Agent run orchestrator.
+
+Drives two agent types:
+
+* **Local directory agent** — sends an ``agent_run`` frame to the connected
+  Electron device, waits for the device to stream back file contents via
+  ``agent_data`` frames, then calls the LLM to extract structured items from
+  each file and pushes inserts to Electron via tool-call round-trips.
+
+* **Cloud connector agent** — fetches data from third-party APIs (Gmail,
+  Teams, Outlook) and pushes extracted items to Electron.  **This path is
+  a stub** — provider integrations are implemented in Step 3.6.
+
+Usage
+-----
+Background tasks are spawned with ``asyncio.create_task()``::
+
+    asyncio.create_task(run_local_agent(user_id, config, run_log, device_manager))
+    asyncio.create_task(trigger_pending_runs(user_id, device_id, device_manager))
+
+The ``trigger_pending_runs`` function is called by the device WS endpoint
+when Electron sends ``device_hello``, so any overdue runs fire immediately
+when the device reconnects.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from croniter import croniter
+from langchain_core.messages import HumanMessage, SystemMessage
+from sqlalchemy import select
+
+from app.core.device_manager import DeviceConnectionManager
+from app.core.llm import get_llm
+from app.db import async_session
+from app.models import AgentRunLog, CloudAgentConfig, LocalAgentConfig
+
+logger = logging.getLogger(__name__)
+
+# ── Timeouts ───────────────────────────────────────────────────────────────
+
+# Max seconds to wait for Electron to finish streaming file data.
+_FILE_READ_TIMEOUT: int = 120
+# Max seconds to wait for Electron to acknowledge a single tool-call insert.
+_INSERT_TIMEOUT: int = 30
+
+# ── Allowed tables & extraction schema hints ───────────────────────────────
+
+_ALLOWED_TABLES: frozenset[str] = frozenset(
+    {"tasks", "notes", "checkpoints", "projects", "taskComments"}
+)
+
+# Field descriptions fed to the extraction LLM as concise schema references.
+_TABLE_SCHEMAS: dict[str, str] = {
+    "tasks": (
+        "title (str, required), description (str), "
+        "status (todo|in_progress|done, default todo), "
+        "priority (high|medium|low, default medium), "
+        "assignee (JSON array string), dueDate (ms timestamp int), projectId (str)"
+    ),
+    "notes": "title (str, required), content (str, markdown), projectId (str)",
+    "checkpoints": (
+        "title (str, required), projectId (str, required), date (ms timestamp int)"
+    ),
+    "projects": "name (str, required), clientId (str)",
+    "taskComments": "taskId (str, required), author (str), content (str, required)",
+}
+
+_EXTRACTION_SYSTEM_PROMPT = """\
+You are a data extraction assistant for a freelance project management tool.
+Given a document, extract structured records matching the user's instructions.
+
+Output a JSON array (no markdown fences, no explanation) of objects shaped:
+  [{{"table": "<table_name>", "data": {{...fields}}}}, ...]
+
+Allowed table names and their fields:
+{table_schemas}
+
+Rules:
+- Only extract tables listed in the "data_types" instructions.
+- Use camelCase field names exactly as shown above.
+- Omit optional fields you cannot determine; do not invent data.
+- Never include id, createdAt, updatedAt, isAiSuggested, or isApproved.
+- If nothing relevant is found, return an empty JSON array: []
+- Return ONLY the JSON array.
+"""
+
+
+# ── Cron helper ────────────────────────────────────────────────────────────
+
+
+def _is_overdue(schedule_cron: str, last_run_at: datetime | None) -> bool:
+    """Return ``True`` if the next scheduled run time has already passed.
+
+    Always validates the cron expression first — an invalid expression returns
+    ``False`` (fail-safe: never trigger an unparseable schedule).
+    """
+    try:
+        now = datetime.now(timezone.utc)
+        if last_run_at is None:
+            # Validate the expression before deciding this is overdue.
+            croniter(schedule_cron, now)
+            return True
+        ts = last_run_at
+        if ts.tzinfo is None:
+            ts = ts.replace(tzinfo=timezone.utc)
+        cron = croniter(schedule_cron, ts)
+        next_run: datetime = cron.get_next(datetime)
+        return now >= next_run
+    except Exception as exc:
+        logger.warning("agent_runner: cannot parse cron %r: %s", schedule_cron, exc)
+        return False  # Fail-safe: don't trigger if expression is invalid.
+
+
+# ── LLM extraction ─────────────────────────────────────────────────────────
+
+
+async def _extract_items_from_content(
+    prompt_template: str,
+    file_content: str,
+    data_types: list[str],
+) -> list[dict[str, Any]]:
+    """Call the LLM to extract structured records from *file_content*.
+
+    Returns a validated list of ``{table: str, data: dict}`` objects.
+    Items referencing tables not in *data_types* are discarded.
+    """
+    allowed = [t for t in data_types if t in _ALLOWED_TABLES]
+    if not allowed:
+        return []
+
+    schema_text = "\n".join(
+        f"  {table}: {_TABLE_SCHEMAS.get(table, '(unknown)')}" for table in allowed
+    )
+    system_prompt = _EXTRACTION_SYSTEM_PROMPT.format(table_schemas=schema_text)
+    user_prompt = (
+        f"User instructions: {prompt_template}\n\n"
+        f"Extract these record types: {', '.join(allowed)}\n\n"
+        f"Document:\n{file_content[:8000]}"
+    )
+
+    llm = get_llm()
+    raw = ""
+    try:
+        response = await llm.ainvoke(
+            [SystemMessage(content=system_prompt), HumanMessage(content=user_prompt)]
+        )
+        raw = str(response.content).strip()
+        items: list[dict] = json.loads(raw)
+        if not isinstance(items, list):
+            raise ValueError("LLM response is not a JSON array")
+    except json.JSONDecodeError as exc:
+        logger.warning(
+            "agent_runner: LLM extraction returned invalid JSON: %s — snippet: %.200r",
+            exc,
+            raw,
+        )
+        return []
+    # Other exceptions (LLM API errors, network errors) propagate to the
+    # caller (run_local_agent) which records them per-file in the run log.
+
+    validated: list[dict[str, Any]] = []
+    for item in items:
+        table = item.get("table")
+        data = item.get("data")
+        if not isinstance(table, str) or table not in allowed:
+            continue
+        if not isinstance(data, dict) or not data:
+            continue
+        # Strip any server-generated or forbidden fields.
+        for _field in ("id", "createdAt", "updatedAt", "isAiSuggested", "isApproved"):
+            data.pop(_field, None)
+        validated.append({"table": table, "data": data})
+    return validated
+
+
+# ── Tool-call insert helper ─────────────────────────────────────────────────
+
+
+async def _send_insert_to_client(
+    user_id: str,
+    table: str,
+    data: dict[str, Any],
+    device_mgr: DeviceConnectionManager,
+) -> dict[str, Any]:
+    """Send an ``insert`` tool_call frame to Electron and await the tool_result.
+
+    All inserts include ``isAiSuggested=1, isApproved=0`` so the user can
+    review AI-produced records before they are treated as confirmed.
+
+    Raises ``asyncio.TimeoutError`` if Electron does not respond within
+    ``_INSERT_TIMEOUT`` seconds.  Raises ``RuntimeError`` if the device
+    disconnects before the frame can be sent.
+    """
+    call_id = str(uuid.uuid4())
+    payload: dict[str, Any] = {
+        "type": "tool_call",
+        "id": call_id,
+        "action": "insert",
+        "table": table,
+        "data": {**data, "isAiSuggested": 1, "isApproved": 0},
+    }
+    fut = device_mgr.create_pending_call(user_id, call_id)
+    await device_mgr.send_frame(user_id, payload)
+    return await asyncio.wait_for(fut, timeout=_INSERT_TIMEOUT)
+
+
+# ── Local agent runner ──────────────────────────────────────────────────────
+
+
+async def run_local_agent(
+    user_id: str,
+    config: LocalAgentConfig,
+    run_log: AgentRunLog,
+    device_mgr: DeviceConnectionManager,
+) -> None:
+    """Execute a local directory agent run end-to-end.
+
+    Steps:
+
+    1. Verify the device identified by ``config.device_id`` is currently online.
+    2. Pre-create the agent_data queue so no incoming frames are lost.
+    3. Send ``agent_run`` frame to Electron (paths, extensions, prompt, data_types).
+    4. Consume ``agent_data`` frames until the ``None`` sentinel from
+       ``agent_complete``.
+    5. For each received file call the LLM to extract ``{table, data}`` items.
+    6. Push each item to Electron as an ``insert`` tool-call; include
+       ``isAiSuggested=1, isApproved=0`` so users can review AI suggestions.
+    7. Persist the run outcome (status, counts, errors) and update
+       ``config.last_run_at``.
+    """
+    run_id = run_log.id
+
+    # ── 1. Device online check ─────────────────────────────────────────
+    if not device_mgr.is_online(user_id, config.device_id):
+        logger.info(
+            "agent_runner: skip run=%s — device %r offline for user=%s",
+            run_id,
+            config.device_id,
+            user_id,
+        )
+        await _finalize_run(
+            run_log,
+            status="error",
+            errors=[f"Device {config.device_id!r} is not connected"],
+        )
+        return
+
+    # ── 2. Pre-create agent_data queue ────────────────────────────────
+    try:
+        device_mgr.get_agent_data_queue(user_id, run_id)
+    except RuntimeError:
+        await _finalize_run(
+            run_log,
+            status="error",
+            errors=["Device disconnected before agent run could start"],
+        )
+        return
+
+    # ── 3. Send agent_run frame ────────────────────────────────────────
+    frame: dict[str, Any] = {
+        "type": "agent_run",
+        "run_id": run_id,
+        "agent_id": config.id,
+        "config": {
+            "paths": config.directory_paths,
+            "file_extensions": config.file_extensions,
+            "prompt_template": config.prompt_template,
+            "data_types": config.data_types,
+        },
+    }
+    try:
+        await device_mgr.send_frame(user_id, frame)
+    except RuntimeError as exc:
+        device_mgr.cleanup_agent_data_queue(user_id, run_id)
+        await _finalize_run(
+            run_log,
+            status="error",
+            errors=[f"Failed to send agent_run frame: {exc}"],
+        )
+        return
+
+    logger.info(
+        "agent_runner: sent agent_run run=%s agent=%s user=%s",
+        run_id,
+        config.id,
+        user_id,
+    )
+
+    # ── 4. Consume agent_data frames ──────────────────────────────────
+    files: list[dict[str, Any]] = []
+    errors: list[str] = []
+
+    try:
+        queue = device_mgr.get_agent_data_queue(user_id, run_id)
+        deadline = asyncio.get_event_loop().time() + _FILE_READ_TIMEOUT
+        while True:
+            remaining = deadline - asyncio.get_event_loop().time()
+            if remaining <= 0:
+                errors.append("Timed out waiting for file data from device")
+                break
+            try:
+                frame_data = await asyncio.wait_for(queue.get(), timeout=remaining)
+            except asyncio.TimeoutError:
+                errors.append("Timed out waiting for file data from device")
+                break
+            if frame_data is None:
+                # Sentinel from agent_complete — stream is done.
+                break
+            files.extend(frame_data.get("files", []))
+    except RuntimeError as exc:
+        errors.append(f"Queue error reading agent data: {exc}")
+
+    # ── 5–6. Extract + insert ─────────────────────────────────────────
+    items_processed = 0
+    items_created = 0
+
+    for file_info in files:
+        file_path: str = file_info.get("path", "<unknown>")
+        content: str = file_info.get("content", "")
+        if not content:
+            continue
+        items_processed += 1
+        try:
+            extracted = await _extract_items_from_content(
+                config.prompt_template, content, config.data_types
+            )
+        except Exception as exc:
+            errors.append(f"LLM extraction error for {file_path!r}: {exc}")
+            continue
+
+        for item in extracted:
+            try:
+                result = await _send_insert_to_client(
+                    user_id, item["table"], item["data"], device_mgr
+                )
+                if result.get("error"):
+                    errors.append(
+                        f"Insert failed ({item['table']}, {file_path!r}): {result['error']}"
+                    )
+                else:
+                    items_created += 1
+            except asyncio.TimeoutError:
+                errors.append(
+                    f"Timed out awaiting insert ack ({item['table']}, {file_path!r})"
+                )
+            except RuntimeError as exc:
+                errors.append(f"Insert error ({item['table']}, {file_path!r}): {exc}")
+
+    # ── 7. Finalise ────────────────────────────────────────────────────
+    device_mgr.cleanup_agent_data_queue(user_id, run_id)
+
+    if errors and items_created == 0:
+        final_status = "error"
+    elif errors:
+        final_status = "partial"
+    else:
+        final_status = "success"
+
+    await _finalize_run(
+        run_log,
+        status=final_status,
+        items_processed=items_processed,
+        items_created=items_created,
+        errors=errors,
+        update_config_last_run=True,
+        config_id=config.id,
+        config_type="local",
+    )
+    logger.info(
+        "agent_runner: run=%s done status=%s processed=%d created=%d errors=%d",
+        run_id,
+        final_status,
+        items_processed,
+        items_created,
+        len(errors),
+    )
+
+
+# ── Cloud agent runner (stub) ───────────────────────────────────────────────
+
+
+async def run_cloud_agent(
+    user_id: str,
+    config: CloudAgentConfig,
+    run_log: AgentRunLog,
+    device_mgr: DeviceConnectionManager,
+) -> None:
+    """Execute a cloud connector agent run.
+
+    .. note::
+        This is a **stub** — provider integrations (Gmail, Teams, Outlook)
+        are implemented in Step 3.6.  The run is immediately marked as an
+        error with an informative message.
+    """
+    logger.info(
+        "agent_runner: cloud agent %s (provider=%s) for user=%s — pending Step 3.6",
+        config.id,
+        config.provider,
+        user_id,
+    )
+    await _finalize_run(
+        run_log,
+        status="error",
+        errors=[
+            f"Cloud provider integrations for '{config.provider}' are not yet "
+            "implemented. This feature arrives in Step 3.6."
+        ],
+    )
+
+
+# ── Pending-run trigger ─────────────────────────────────────────────────────
+
+
+async def trigger_pending_runs(
+    user_id: str,
+    device_id: str,
+    device_mgr: DeviceConnectionManager,
+) -> None:
+    """Dispatch any overdue agent runs after an Electron device connects.
+
+    Called as a background task from the device WS endpoint on ``device_hello``.
+
+    Scheduling rules:
+
+    * **Local agents**: only triggered when ``config.device_id == device_id``.
+    * **Cloud agents**: triggered on any connected device (no device binding).
+    * Runs execute **sequentially** to avoid flooding the WS connection.
+    """
+    logger.info(
+        "agent_runner: scanning overdue runs for user=%s device=%s", user_id, device_id
+    )
+    async with async_session() as db:
+        local_result = await db.execute(
+            select(LocalAgentConfig).where(
+                LocalAgentConfig.user_id == user_id,
+                LocalAgentConfig.enabled == True,  # noqa: E712
+                LocalAgentConfig.device_id == device_id,
+            )
+        )
+        local_configs: list[LocalAgentConfig] = list(local_result.scalars().all())
+
+        cloud_result = await db.execute(
+            select(CloudAgentConfig).where(
+                CloudAgentConfig.user_id == user_id,
+                CloudAgentConfig.enabled == True,  # noqa: E712
+            )
+        )
+        cloud_configs: list[CloudAgentConfig] = list(cloud_result.scalars().all())
+
+    # Build ordered list of overdue (type, config) pairs.
+    pending: list[tuple[str, Any]] = []
+    for cfg in local_configs:
+        if _is_overdue(cfg.schedule_cron, cfg.last_run_at):
+            pending.append(("local", cfg))
+    for cfg in cloud_configs:
+        if _is_overdue(cfg.schedule_cron, cfg.last_run_at):
+            pending.append(("cloud", cfg))
+
+    if not pending:
+        logger.debug("agent_runner: no overdue runs for user=%s", user_id)
+        return
+
+    logger.info(
+        "agent_runner: %d overdue run(s) to dispatch for user=%s", len(pending), user_id
+    )
+
+    for agent_type, cfg in pending:
+        # Create a fresh run log for this scheduled dispatch.
+        run_log = AgentRunLog(
+            agent_id=cfg.id,
+            agent_type=agent_type,
+            user_id=user_id,
+            status="running",
+        )
+        async with async_session() as db:
+            db.add(run_log)
+            await db.commit()
+            await db.refresh(run_log)
+
+        if agent_type == "local":
+            await run_local_agent(user_id, cfg, run_log, device_mgr)
+        else:
+            await run_cloud_agent(user_id, cfg, run_log, device_mgr)
+
+
+# ── Internal helper ─────────────────────────────────────────────────────────
+
+
+async def _finalize_run(
+    run_log: AgentRunLog,
+    *,
+    status: str,
+    items_processed: int = 0,
+    items_created: int = 0,
+    errors: list[str] | None = None,
+    update_config_last_run: bool = False,
+    config_id: str | None = None,
+    config_type: str | None = None,
+) -> None:
+    """Persist the run outcome and optionally update ``LocalAgentConfig.last_run_at``.
+
+    Uses a fresh DB session so this is safe to call from background tasks
+    after the original request session has closed.
+    """
+    now = datetime.now(timezone.utc)
+    try:
+        async with async_session() as db:
+            managed = await db.merge(run_log)
+            managed.status = status
+            managed.items_processed = items_processed
+            managed.items_created = items_created
+            managed.errors = errors or []
+            managed.completed_at = now
+
+            if update_config_last_run and config_id and config_type == "local":
+                cfg_result = await db.execute(
+                    select(LocalAgentConfig).where(LocalAgentConfig.id == config_id)
+                )
+                cfg = cfg_result.scalar_one_or_none()
+                if cfg:
+                    cfg.last_run_at = now
+
+            await db.commit()
+    except Exception as exc:
+        logger.error(
+            "agent_runner: failed to finalize run_log=%s: %s", run_log.id, exc
+        )

app/core/device_manager.py

@@ -0,0 +1,183 @@
+"""Device connection manager.
+
+Maintains in-memory state for all active Electron → backend WebSocket
+connections.  One connection per user (latest replaces previous).
+
+The manager participates in two interaction patterns:
+
+1. **Tool-call round-trip** (bidirectional CRUD):
+   - Backend sends ``tool_call`` frame → Electron executes CRUD → returns
+     ``tool_result`` frame.
+   - ``create_pending_call`` registers a Future keyed by ``call_id``.
+   - ``resolve_pending_call`` fulfils the Future; callers awaiting it
+     receive the result dict from Electron.
+
+2. **Agent-data streaming** (local directory agent runs):
+   - Backend sends ``agent_run`` frame → Electron reads files and sends
+     back a stream of ``agent_data`` frames followed by ``agent_complete``.
+   - ``get_agent_data_queue`` returns (or creates) an asyncio.Queue for
+     a specific ``run_id`` so the agent runner can iterate frames.
+
+The ``device_manager`` module-level singleton is imported by both the
+device WS route and the agent runner.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from dataclasses import dataclass, field
+
+from fastapi import WebSocket
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class DeviceConnection:
+    """State for a single connected Electron device."""
+
+    ws: WebSocket
+    device_id: str
+    # Futures indexed by tool_call id — resolved when tool_result arrives.
+    pending_calls: dict[str, asyncio.Future[dict]] = field(default_factory=dict)
+    # Per-run queues for agent_data / agent_complete frames.
+    agent_data_queues: dict[str, asyncio.Queue[dict | None]] = field(default_factory=dict)
+
+
+class DeviceConnectionManager:
+    """Singleton registry of active Electron WebSocket connections.
+
+    Thread/task safety note: asyncio is single-threaded by design.  All
+    mutations happen inside await-points on the main event loop, so no
+    locking is required for the in-memory dicts.
+    """
+
+    def __init__(self) -> None:
+        self._connections: dict[str, DeviceConnection] = {}
+
+    # ── Registration ──────────────────────────────────────────────────
+
+    def register(self, user_id: str, device_id: str, ws: WebSocket) -> None:
+        """Store the active connection for *user_id*, replacing any previous one."""
+        if user_id in self._connections:
+            old = self._connections[user_id]
+            logger.info(
+                "device_manager: replacing existing connection for user=%s device=%s",
+                user_id,
+                old.device_id,
+            )
+            # Cancel any futures that were waiting on the old connection.
+            for fut in old.pending_calls.values():
+                if not fut.done():
+                    fut.cancel()
+        self._connections[user_id] = DeviceConnection(ws=ws, device_id=device_id)
+        logger.info(
+            "device_manager: registered user=%s device=%s", user_id, device_id
+        )
+
+    def unregister(self, user_id: str) -> None:
+        """Remove the connection for *user_id* and cancel any pending futures."""
+        conn = self._connections.pop(user_id, None)
+        if conn is None:
+            return
+        for fut in conn.pending_calls.values():
+            if not fut.done():
+                fut.cancel()
+        logger.info("device_manager: unregistered user=%s", user_id)
+
+    # ── Presence queries ──────────────────────────────────────────────
+
+    def get_ws(self, user_id: str) -> WebSocket | None:
+        """Return the active WebSocket for *user_id*, or ``None`` if offline."""
+        conn = self._connections.get(user_id)
+        return conn.ws if conn else None
+
+    def is_online(self, user_id: str, device_id: str | None = None) -> bool:
+        """Return ``True`` if the user has an active connection.
+
+        If *device_id* is provided also checks that it matches the connected device.
+        """
+        conn = self._connections.get(user_id)
+        if conn is None:
+            return False
+        if device_id is not None:
+            return conn.device_id == device_id
+        return True
+
+    # ── Frame sending ─────────────────────────────────────────────────
+
+    async def send_frame(self, user_id: str, frame: dict) -> None:
+        """Send *frame* as a JSON text message to the device.
+
+        Raises ``RuntimeError`` if the user is not connected.
+        """
+        conn = self._connections.get(user_id)
+        if conn is None:
+            raise RuntimeError(
+                f"send_frame: user {user_id!r} is not connected"
+            )
+        await conn.ws.send_text(json.dumps(frame))
+
+    # ── Tool-call round-trip ──────────────────────────────────────────
+
+    def create_pending_call(
+        self, user_id: str, call_id: str
+    ) -> asyncio.Future[dict]:
+        """Register a Future that will be resolved when the tool_result arrives.
+
+        Raises ``RuntimeError`` if the user is not connected.
+        """
+        conn = self._connections.get(user_id)
+        if conn is None:
+            raise RuntimeError(
+                f"create_pending_call: user {user_id!r} is not connected"
+            )
+        loop = asyncio.get_event_loop()
+        fut: asyncio.Future[dict] = loop.create_future()
+        conn.pending_calls[call_id] = fut
+        return fut
+
+    def resolve_pending_call(
+        self, user_id: str, call_id: str, result: dict
+    ) -> None:
+        """Fulfil the Future registered under *call_id* with the Electron result.
+
+        No-ops if the call_id is unknown (already timed out or cancelled).
+        """
+        conn = self._connections.get(user_id)
+        if conn is None:
+            return
+        fut = conn.pending_calls.pop(call_id, None)
+        if fut is not None and not fut.done():
+            fut.set_result(result)
+
+    # ── Agent-data queue ──────────────────────────────────────────────
+
+    def get_agent_data_queue(
+        self, user_id: str, run_id: str
+    ) -> asyncio.Queue[dict | None]:
+        """Return (creating if absent) the queue for *run_id* agent frames.
+
+        The agent runner reads from this queue.  The device WS handler writes
+        to it.  ``None`` is the sentinel that signals the stream is finished.
+        """
+        conn = self._connections.get(user_id)
+        if conn is None:
+            raise RuntimeError(
+                f"get_agent_data_queue: user {user_id!r} is not connected"
+            )
+        if run_id not in conn.agent_data_queues:
+            conn.agent_data_queues[run_id] = asyncio.Queue()
+        return conn.agent_data_queues[run_id]
+
+    def cleanup_agent_data_queue(self, user_id: str, run_id: str) -> None:
+        """Remove the queue for *run_id* once a run has completed."""
+        conn = self._connections.get(user_id)
+        if conn:
+            conn.agent_data_queues.pop(run_id, None)
+
+
+# Module-level singleton — import this everywhere.
+device_manager = DeviceConnectionManager()

app/main.py

@@ -43,7 +43,7 @@ def create_app() -> FastAPI:
     app.add_middleware(SanitizerMiddleware)
     app.add_middleware(TierRateLimitMiddleware)
 
-    from app.api.routes import auth, backup, billing, chat, plans, plugins, storage, vectors
+    from app.api.routes import agents, auth, backup, billing, chat, device_ws, plans, plugins, storage, vectors
 
     app.include_router(auth.router,       prefix="/api/v1")
     app.include_router(chat.router,       prefix="/api/v1")
@@ -53,6 +53,8 @@ def create_app() -> FastAPI:
     app.include_router(backup.router,     prefix="/api/v1")
     app.include_router(plugins.router,    prefix="/api/v1")
     app.include_router(billing.router,    prefix="/api/v1")
+    app.include_router(agents.router,     prefix="/api/v1")
+    app.include_router(device_ws.router,  prefix="/api/v1")
 
     @app.get("/api/v1/health", tags=["health"])
     async def health() -> dict:

app/models.py

@@ -23,11 +23,13 @@ from datetime import datetime, timezone
 
 from sqlalchemy import (
     BigInteger,
+    Boolean,
     DateTime,
     Enum,
     Float,
     ForeignKey,
     Integer,
+    JSON,
     String,
     Text,
     UniqueConstraint,
@@ -54,6 +56,9 @@ def _now() -> datetime:
 TierEnum = Enum("free", "pro", "power", "team", name="billing_tier")
 PluginStatusEnum = Enum("pending_review", "approved", "rejected", name="plugin_status")
 ReviewDecisionEnum = Enum("approved", "rejected", name="review_decision")
+AgentTypeEnum = Enum("local", "cloud", name="agent_type")
+AgentStatusEnum = Enum("running", "success", "error", "partial", name="agent_run_status")
+CloudProviderEnum = Enum("gmail", "teams", "outlook", name="cloud_provider")
 
 
 # ── Models ────────────────────────────────────────────────────────────────
@@ -266,3 +271,107 @@ class RevenueEvent(Base):
     )
 
     plugin: Mapped[Plugin] = relationship(back_populates="revenue_events")
+
+
+class LocalAgentConfig(Base):
+    __tablename__ = "local_agent_configs"
+
+    id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), primary_key=True, default=_uuid
+    )
+    user_id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    device_id: Mapped[str] = mapped_column(String(255), nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    directory_paths: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
+    data_types: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
+    prompt_template: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    file_extensions: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
+    schedule_cron: Mapped[str] = mapped_column(String(100), nullable=False, default="0 */6 * * *")
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+    run_logs: Mapped[list[AgentRunLog]] = relationship(
+        back_populates="local_agent",
+        primaryjoin="and_(AgentRunLog.agent_id == LocalAgentConfig.id, AgentRunLog.agent_type == 'local')",
+        foreign_keys="AgentRunLog.agent_id",
+        cascade="all, delete-orphan",
+        overlaps="run_logs,cloud_agent",
+    )
+
+
+class CloudAgentConfig(Base):
+    __tablename__ = "cloud_agent_configs"
+
+    id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), primary_key=True, default=_uuid
+    )
+    user_id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    provider: Mapped[str] = mapped_column(CloudProviderEnum, nullable=False)
+    name: Mapped[str] = mapped_column(String(255), nullable=False)
+    data_types: Mapped[list] = mapped_column(JSON, nullable=False, default=list)
+    prompt_template: Mapped[str] = mapped_column(Text, nullable=False, default="")
+    oauth_token_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    filter_config: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    schedule_cron: Mapped[str] = mapped_column(String(100), nullable=False, default="0 */6 * * *")
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now()
+    )
+
+    run_logs: Mapped[list[AgentRunLog]] = relationship(
+        back_populates="cloud_agent",
+        primaryjoin="and_(AgentRunLog.agent_id == CloudAgentConfig.id, AgentRunLog.agent_type == 'cloud')",
+        foreign_keys="AgentRunLog.agent_id",
+        cascade="all, delete-orphan",
+        overlaps="run_logs,local_agent",
+    )
+
+
+class AgentRunLog(Base):
+    __tablename__ = "agent_run_logs"
+
+    id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), primary_key=True, default=_uuid
+    )
+    # Plain string — not a FK because it references either local_agent_configs or cloud_agent_configs
+    # depending on agent_type. Query by (agent_id, agent_type) to locate the source config.
+    agent_id: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
+    agent_type: Mapped[str] = mapped_column(AgentTypeEnum, nullable=False)
+    user_id: Mapped[str] = mapped_column(
+        Uuid(as_uuid=False), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    status: Mapped[str] = mapped_column(AgentStatusEnum, nullable=False, default="running")
+    items_processed: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    items_created: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    errors: Mapped[list | None] = mapped_column(JSON, nullable=True)
+    started_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    local_agent: Mapped[LocalAgentConfig | None] = relationship(
+        back_populates="run_logs",
+        primaryjoin="and_(AgentRunLog.agent_id == LocalAgentConfig.id, AgentRunLog.agent_type == 'local')",
+        foreign_keys="AgentRunLog.agent_id",
+        overlaps="run_logs,cloud_agent",
+    )
+    cloud_agent: Mapped[CloudAgentConfig | None] = relationship(
+        back_populates="run_logs",
+        primaryjoin="and_(AgentRunLog.agent_id == CloudAgentConfig.id, AgentRunLog.agent_type == 'cloud')",
+        foreign_keys="AgentRunLog.agent_id",
+        overlaps="run_logs,local_agent",
+    )

app/schemas.py

@@ -167,6 +167,10 @@ class WsFrameType(str, Enum):
     tool_result = "tool_result"
     final = "final"
     ping = "ping"
+    agent_run = "agent_run"
+    agent_data = "agent_data"
+    agent_complete = "agent_complete"
+    device_hello = "device_hello"
 
 
 class WsToolCall(BaseModel):
@@ -207,3 +211,139 @@ class WsFinal(BaseModel):
 
     type: Literal[WsFrameType.final] = WsFrameType.final
     response: str
+
+
+# ── WebSocket Agent Frame Protocol ────────────────────────────────────
+
+class WsDeviceHello(BaseModel):
+    """Client → Server: device identification on WS connect."""
+
+    type: Literal[WsFrameType.device_hello] = WsFrameType.device_hello
+    device_id: str
+    agent_ids: list[str] = Field(default_factory=list)
+
+
+class WsAgentRun(BaseModel):
+    """Server → Client: trigger an agent run on the connected device."""
+
+    type: Literal[WsFrameType.agent_run] = WsFrameType.agent_run
+    run_id: str
+    agent_id: str
+    config: dict[str, Any]
+
+
+class WsAgentData(BaseModel):
+    """Client → Server: files read by the local agent."""
+
+    type: Literal[WsFrameType.agent_data] = WsFrameType.agent_data
+    run_id: str
+    files: list[dict[str, Any]]
+
+
+class WsAgentComplete(BaseModel):
+    """Client → Server: Electron signals it has finished reading files."""
+
+    type: Literal[WsFrameType.agent_complete] = WsFrameType.agent_complete
+    run_id: str
+    files_read: int
+    errors: list[str] = Field(default_factory=list)
+
+
+# ── Agent Catalog ─────────────────────────────────────────────────────
+
+class AgentCatalogItem(BaseModel):
+    type: str
+    name: str
+    description: str
+    config_schema: dict[str, Any] = Field(default_factory=dict)
+
+
+# ── Local Agent Config ────────────────────────────────────────────────
+
+class LocalAgentConfigCreate(BaseModel):
+    name: str
+    device_id: str
+    directory_paths: list[str]
+    data_types: list[str]
+    prompt_template: str
+    file_extensions: list[str]
+    schedule_cron: str
+
+
+class LocalAgentConfigUpdate(BaseModel):
+    name: str | None = None
+    device_id: str | None = None
+    directory_paths: list[str] | None = None
+    data_types: list[str] | None = None
+    prompt_template: str | None = None
+    file_extensions: list[str] | None = None
+    schedule_cron: str | None = None
+    enabled: bool | None = None
+
+
+class LocalAgentConfigResponse(BaseModel):
+    id: str
+    name: str
+    device_id: str
+    directory_paths: list[str]
+    data_types: list[str]
+    prompt_template: str
+    file_extensions: list[str]
+    schedule_cron: str
+    enabled: bool
+    last_run_at: int | None
+    created_at: int
+    updated_at: int
+
+
+# ── Cloud Agent Config ────────────────────────────────────────────────
+
+class CloudAgentConfigCreate(BaseModel):
+    provider: Literal["gmail", "teams", "outlook"]
+    name: str
+    data_types: list[str]
+    prompt_template: str
+    oauth_token_encrypted: str
+    schedule_cron: str
+    filter_config: dict[str, Any] | None = None
+
+
+class CloudAgentConfigUpdate(BaseModel):
+    provider: Literal["gmail", "teams", "outlook"] | None = None
+    name: str | None = None
+    data_types: list[str] | None = None
+    prompt_template: str | None = None
+    oauth_token_encrypted: str | None = None
+    schedule_cron: str | None = None
+    filter_config: dict[str, Any] | None = None
+    enabled: bool | None = None
+
+
+class CloudAgentConfigResponse(BaseModel):
+    """oauth_token_encrypted is intentionally excluded — never returned to clients."""
+
+    id: str
+    provider: Literal["gmail", "teams", "outlook"]
+    name: str
+    data_types: list[str]
+    prompt_template: str
+    schedule_cron: str
+    filter_config: dict[str, Any] | None
+    enabled: bool
+    last_run_at: int | None
+    created_at: int
+    updated_at: int
+
+
+# ── Agent Run Log ─────────────────────────────────────────────────────
+
+class AgentRunLogResponse(BaseModel):
+    id: str
+    agent_id: str
+    agent_type: Literal["local", "cloud"]
+    status: Literal["running", "success", "error", "partial"]
+    items_processed: int
+    items_created: int
+    errors: list[str]
+    started_at: int
+    completed_at: int | None

requirements.txt

@@ -24,4 +24,5 @@ aiosqlite>=0.20.0
 moto[s3]>=5.0.0
 pinecone>=5.0.0
 qdrant-client>=1.7.0
+croniter>=3.0.0
 ruff>=0.8.0

tests/test_agent_runner.py

@@ -0,0 +1,660 @@
+"""Tests for Step 3.4: agent_runner module.
+
+Coverage:
+  Unit:
+    - _is_overdue      — cron schedule overdue detection
+    - _extract_items_from_content — LLM extraction + JSON parsing + validation
+    - _send_insert_to_client      — tool_call frame construction + timeout
+    - run_local_agent             — end-to-end local agent happy path
+    - run_local_agent             — device offline path
+    - run_local_agent             — file-read timeout path
+    - run_local_agent             — LLM extraction error path
+    - run_cloud_agent             — stub returns error immediately
+    - trigger_pending_runs        — overdue local + cloud dispatched
+    - trigger_pending_runs        — non-overdue skipped
+    - trigger_pending_runs        — device_id filter for local agents
+
+  Integration:
+    - POST /agents/{id}/run       — 404 on unknown agent
+    - POST /agents/{id}/run       — creates run log + dispatches background task
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import pytest_asyncio
+
+from app.core.agent_runner import (
+    _extract_items_from_content,
+    _is_overdue,
+    _send_insert_to_client,
+    run_cloud_agent,
+    run_local_agent,
+    trigger_pending_runs,
+)
+from app.core.device_manager import DeviceConnectionManager
+from app.db import get_session
+from app.main import app
+from app.models import AgentRunLog, CloudAgentConfig, LocalAgentConfig
+from tests.conftest import TEST_USER_IDS, auth_header
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_FREE_UID = TEST_USER_IDS["free"]
+_PRO_UID = TEST_USER_IDS["pro"]
+
+
+def _make_local_config(user_id: str = _FREE_UID, device_id: str = "dev-001") -> LocalAgentConfig:
+    return LocalAgentConfig(
+        id=str(uuid.uuid4()),
+        user_id=user_id,
+        device_id=device_id,
+        name="Test Local Agent",
+        directory_paths=["/home/user/emails"],
+        data_types=["tasks", "notes"],
+        prompt_template="Extract tasks and notes from this document.",
+        file_extensions=[".txt", ".eml"],
+        schedule_cron="0 */6 * * *",
+        enabled=True,
+        last_run_at=None,
+    )
+
+
+def _make_cloud_config(user_id: str = _FREE_UID) -> CloudAgentConfig:
+    return CloudAgentConfig(
+        id=str(uuid.uuid4()),
+        user_id=user_id,
+        provider="gmail",
+        name="Test Gmail Agent",
+        data_types=["tasks"],
+        prompt_template="Extract tasks from email.",
+        schedule_cron="0 */6 * * *",
+        enabled=True,
+        last_run_at=None,
+    )
+
+
+def _make_run_log(agent_id: str, agent_type: str = "local", user_id: str = _FREE_UID) -> AgentRunLog:
+    return AgentRunLog(
+        id=str(uuid.uuid4()),
+        agent_id=agent_id,
+        agent_type=agent_type,
+        user_id=user_id,
+        status="running",
+        started_at=datetime.now(timezone.utc),
+    )
+
+
+def _make_manager(user_id: str = _FREE_UID, device_id: str = "dev-001") -> DeviceConnectionManager:
+    mgr = DeviceConnectionManager()
+    ws = MagicMock()
+    ws.send_text = AsyncMock()
+    mgr.register(user_id, device_id, ws)
+    return mgr
+
+
+# ---------------------------------------------------------------------------
+# _is_overdue
+# ---------------------------------------------------------------------------
+
+def test_is_overdue_never_run():
+    """An agent that has never run is always overdue."""
+    assert _is_overdue("0 */6 * * *", None) is True
+
+
+def test_is_overdue_very_recently_run():
+    """An agent that just ran is not overdue."""
+    last = datetime.now(timezone.utc)
+    assert _is_overdue("0 */6 * * *", last) is False
+
+
+def test_is_overdue_long_ago():
+    """An agent last run 2 days ago with a 6-hour schedule is overdue."""
+    from datetime import timedelta
+    last = datetime.now(timezone.utc) - timedelta(days=2)
+    assert _is_overdue("0 */6 * * *", last) is True
+
+
+def test_is_overdue_invalid_cron_returns_false():
+    """Unparseable cron must not raise and should return False (fail-safe)."""
+    assert _is_overdue("not a cron", None) is False
+
+
+def test_is_overdue_naive_datetime():
+    """Naive datetime objects are handled without raising."""
+    from datetime import timedelta
+    last = datetime.utcnow() - timedelta(days=1)  # naive
+    # Should not raise.
+    result = _is_overdue("0 */6 * * *", last)
+    assert isinstance(result, bool)
+
+
+# ---------------------------------------------------------------------------
+# _extract_items_from_content
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_extract_items_happy_path():
+    """LLM returns valid JSON array; items with allowed tables are returned."""
+    mock_llm = MagicMock()
+    mock_response = MagicMock()
+    mock_response.content = json.dumps([
+        {"table": "tasks", "data": {"title": "Buy milk", "priority": "high"}},
+        {"table": "notes", "data": {"title": "Meeting recap", "content": "Discussed roadmap"}},
+    ])
+    mock_llm.ainvoke = AsyncMock(return_value=mock_response)
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm):
+        items = await _extract_items_from_content(
+            "Extract tasks and notes.",
+            "Email body: Buy milk urgently. Notes from meeting: discussed roadmap.",
+            ["tasks", "notes"],
+        )
+
+    assert len(items) == 2
+    assert items[0]["table"] == "tasks"
+    assert items[0]["data"]["title"] == "Buy milk"
+    assert items[1]["table"] == "notes"
+
+
+@pytest.mark.asyncio
+async def test_extract_items_strips_forbidden_fields():
+    """Fields like id, createdAt, isAiSuggested must be stripped from extracted data."""
+    mock_llm = MagicMock()
+    mock_response = MagicMock()
+    mock_response.content = json.dumps([
+        {
+            "table": "tasks",
+            "data": {
+                "title": "Review PR",
+                "id": "should-be-removed",
+                "createdAt": 99999,
+                "isAiSuggested": 0,
+                "isApproved": 1,
+            },
+        }
+    ])
+    mock_llm.ainvoke = AsyncMock(return_value=mock_response)
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm):
+        items = await _extract_items_from_content("Extract tasks.", "Review the PR.", ["tasks"])
+
+    assert len(items) == 1
+    data = items[0]["data"]
+    assert "id" not in data
+    assert "createdAt" not in data
+    assert "isAiSuggested" not in data
+    assert "isApproved" not in data
+    assert data["title"] == "Review PR"
+
+
+@pytest.mark.asyncio
+async def test_extract_items_invalid_json_returns_empty():
+    """LLM returning invalid JSON must return empty list without raising."""
+    mock_llm = MagicMock()
+    mock_response = MagicMock()
+    mock_response.content = "Sorry, I cannot extract anything."
+    mock_llm.ainvoke = AsyncMock(return_value=mock_response)
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm):
+        items = await _extract_items_from_content("Extract tasks.", "content", ["tasks"])
+
+    assert items == []
+
+
+@pytest.mark.asyncio
+async def test_extract_items_disallowed_table_filtered():
+    """Items whose table is not in data_types are discarded."""
+    mock_llm = MagicMock()
+    mock_response = MagicMock()
+    mock_response.content = json.dumps([
+        {"table": "tasks", "data": {"title": "Valid task"}},
+        {"table": "projects", "data": {"name": "Should be filtered"}},
+    ])
+    mock_llm.ainvoke = AsyncMock(return_value=mock_response)
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm):
+        # Only "tasks" is in data_types — "projects" should be filtered.
+        items = await _extract_items_from_content("Extract.", "content", ["tasks"])
+
+    assert len(items) == 1
+    assert items[0]["table"] == "tasks"
+
+
+@pytest.mark.asyncio
+async def test_extract_items_empty_data_types_returns_empty():
+    """If no allowed data_types match, skip LLM call and return immediately."""
+    mock_llm = MagicMock()
+    mock_llm.ainvoke = AsyncMock()
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm):
+        items = await _extract_items_from_content("Extract.", "content", [])
+
+    mock_llm.ainvoke.assert_not_called()
+    assert items == []
+
+
+@pytest.mark.asyncio
+async def test_extract_items_llm_error_propagates():
+    """LLM API errors propagate so the caller (run_local_agent) can record them."""
+    mock_llm = MagicMock()
+    mock_llm.ainvoke = AsyncMock(side_effect=RuntimeError("API unavailable"))
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm):
+        with pytest.raises(RuntimeError, match="API unavailable"):
+            await _extract_items_from_content("Extract tasks.", "content", ["tasks"])
+
+
+# ---------------------------------------------------------------------------
+# _send_insert_to_client
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_send_insert_to_client_happy_path():
+    """Frame is sent with isAiSuggested/isApproved added; result is returned."""
+    mgr = _make_manager()
+
+    sent_payloads: list[dict] = []
+    original_send = mgr.send_frame
+
+    async def _capture_send(uid: str, frame: dict) -> None:
+        sent_payloads.append(frame)
+        # Immediately resolve the pending call with a success result.
+        call_id = frame["id"]
+        mgr.resolve_pending_call(uid, call_id, {"row": {"id": "new-id", "title": "Buy milk"}})
+
+    mgr.send_frame = _capture_send  # type: ignore[method-assign]
+
+    result = await _send_insert_to_client(
+        _FREE_UID, "tasks", {"title": "Buy milk", "priority": "high"}, mgr
+    )
+
+    assert len(sent_payloads) == 1
+    payload = sent_payloads[0]
+    assert payload["action"] == "insert"
+    assert payload["table"] == "tasks"
+    assert payload["data"]["title"] == "Buy milk"
+    assert payload["data"]["isAiSuggested"] == 1
+    assert payload["data"]["isApproved"] == 0
+    assert result["row"]["title"] == "Buy milk"
+
+
+@pytest.mark.asyncio
+async def test_send_insert_to_client_timeout():
+    """asyncio.TimeoutError is raised when Electron does not respond."""
+    mgr = _make_manager()
+
+    async def _slow_send(uid: str, frame: dict) -> None:
+        # Never resolve the pending call.
+        pass
+
+    mgr.send_frame = _slow_send  # type: ignore[method-assign]
+
+    with patch("app.core.agent_runner._INSERT_TIMEOUT", 0.05):
+        with pytest.raises(asyncio.TimeoutError):
+            await _send_insert_to_client(_FREE_UID, "tasks", {"title": "X"}, mgr)
+
+
+# ---------------------------------------------------------------------------
+# run_local_agent
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_local_agent_device_offline():
+    """run_local_agent marks run as error when device is offline."""
+    config = _make_local_config()
+    run_log = _make_run_log(config.id)
+    mgr = DeviceConnectionManager()  # Empty — no device registered.
+
+    with patch("app.core.agent_runner._finalize_run", new_callable=AsyncMock) as mock_finalize:
+        await run_local_agent(_FREE_UID, config, run_log, mgr)
+
+    mock_finalize.assert_called_once()
+    _args, kwargs = mock_finalize.call_args
+    assert kwargs["status"] == "error"
+    assert any("not connected" in e for e in kwargs["errors"])
+
+
+@pytest.mark.asyncio
+async def test_run_local_agent_happy_path():
+    """End-to-end: files received, LLM extracts one task, insert sent + ack'd."""
+    config = _make_local_config()
+    run_log = _make_run_log(config.id)
+    mgr = _make_manager()
+
+    # Build a fake agent_data frame (will be queued after send).
+    file_frame = {
+        "type": "agent_data",
+        "run_id": run_log.id,
+        "files": [{"path": "/email.eml", "content": "Urgent: fix the bug by Friday."}],
+    }
+    agent_complete_frame = None  # sentinel
+
+    sent_frames: list[dict] = []
+
+    async def _mock_send(uid: str, frame: dict) -> None:
+        sent_frames.append(frame)
+        if frame.get("type") == "agent_run":
+            # Simulate Electron responding with file data then agent_complete.
+            q = mgr.get_agent_data_queue(uid, frame["run_id"])
+            await q.put(file_frame)
+            await q.put(agent_complete_frame)
+        elif frame.get("type") == "tool_call":
+            # Resolve the pending insert immediately.
+            mgr.resolve_pending_call(uid, frame["id"], {"row": {"id": "new-task", "title": "Fix the bug"}})
+
+    mgr.send_frame = _mock_send  # type: ignore[method-assign]
+
+    mock_llm = MagicMock()
+    mock_response = MagicMock()
+    mock_response.content = json.dumps([
+        {"table": "tasks", "data": {"title": "Fix the bug", "priority": "high"}}
+    ])
+    mock_llm.ainvoke = AsyncMock(return_value=mock_response)
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm), \
+         patch("app.core.agent_runner._finalize_run", new_callable=AsyncMock) as mock_finalize:
+        await run_local_agent(_FREE_UID, config, run_log, mgr)
+
+    mock_finalize.assert_called_once()
+    _args, kwargs = mock_finalize.call_args
+    assert kwargs["status"] == "success"
+    assert kwargs["items_processed"] == 1
+    assert kwargs["items_created"] == 1
+    assert kwargs["errors"] == []
+    assert kwargs["update_config_last_run"] is True
+
+    # Verify agent_run frame was sent.
+    agent_run_frames = [f for f in sent_frames if f.get("type") == "agent_run"]
+    assert len(agent_run_frames) == 1
+    assert agent_run_frames[0]["agent_id"] == config.id
+    assert "paths" in agent_run_frames[0]["config"]
+
+    # Verify insert frame was sent with AI flags.
+    insert_frames = [f for f in sent_frames if f.get("type") == "tool_call"]
+    assert len(insert_frames) == 1
+    assert insert_frames[0]["data"]["isAiSuggested"] == 1
+    assert insert_frames[0]["data"]["isApproved"] == 0
+
+
+@pytest.mark.asyncio
+async def test_run_local_agent_file_read_timeout():
+    """run_local_agent marks run as partial/error when device stops sending files."""
+    config = _make_local_config()
+    run_log = _make_run_log(config.id)
+    mgr = _make_manager()
+
+    async def _mock_send(uid: str, frame: dict) -> None:
+        # Don't put anything in the queue — simulate stalled device.
+        pass
+
+    mgr.send_frame = _mock_send  # type: ignore[method-assign]
+
+    with patch("app.core.agent_runner._FILE_READ_TIMEOUT", 0.1), \
+         patch("app.core.agent_runner._finalize_run", new_callable=AsyncMock) as mock_finalize:
+        await run_local_agent(_FREE_UID, config, run_log, mgr)
+
+    mock_finalize.assert_called_once()
+    _args, kwargs = mock_finalize.call_args
+    assert kwargs["status"] == "error"  # No items created, so error (not partial).
+    assert any("timed out" in e.lower() for e in kwargs["errors"])
+
+
+@pytest.mark.asyncio
+async def test_run_local_agent_llm_extraction_error():
+    """LLM errors per-file are recorded; run continues for remaining files."""
+    config = _make_local_config()
+    run_log = _make_run_log(config.id)
+    mgr = _make_manager()
+
+    file_frame = {
+        "type": "agent_data",
+        "run_id": run_log.id,
+        "files": [
+            {"path": "/file1.eml", "content": "Email one."},
+            {"path": "/file2.eml", "content": "Email two."},
+        ],
+    }
+
+    async def _mock_send(uid: str, frame: dict) -> None:
+        if frame.get("type") == "agent_run":
+            q = mgr.get_agent_data_queue(uid, frame["run_id"])
+            await q.put(file_frame)
+            await q.put(None)  # agent_complete sentinel
+
+    mgr.send_frame = _mock_send  # type: ignore[method-assign]
+
+    mock_llm = MagicMock()
+    mock_llm.ainvoke = AsyncMock(side_effect=RuntimeError("LLM boom"))
+
+    with patch("app.core.agent_runner.get_llm", return_value=mock_llm), \
+         patch("app.core.agent_runner._finalize_run", new_callable=AsyncMock) as mock_finalize:
+        await run_local_agent(_FREE_UID, config, run_log, mgr)
+
+    _args, kwargs = mock_finalize.call_args
+    assert kwargs["status"] == "error"
+    assert kwargs["items_processed"] == 2  # Both files attempted.
+    assert kwargs["items_created"] == 0
+    assert len(kwargs["errors"]) == 2  # One error per file.
+
+
+# ---------------------------------------------------------------------------
+# run_cloud_agent (stub)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_run_cloud_agent_stub_returns_error():
+    """Cloud agent stub immediately marks run as error with informative message."""
+    config = _make_cloud_config()
+    run_log = _make_run_log(config.id, agent_type="cloud")
+    mgr = _make_manager()
+
+    with patch("app.core.agent_runner._finalize_run", new_callable=AsyncMock) as mock_finalize:
+        await run_cloud_agent(_FREE_UID, config, run_log, mgr)
+
+    mock_finalize.assert_called_once()
+    _args, kwargs = mock_finalize.call_args
+    assert kwargs["status"] == "error"
+    assert len(kwargs["errors"]) == 1
+    assert "gmail" in kwargs["errors"][0].lower()
+    assert "3.6" in kwargs["errors"][0]
+
+
+# ---------------------------------------------------------------------------
+# trigger_pending_runs
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_trigger_pending_runs_no_overdue():
+    """If no agents are overdue trigger_pending_runs does nothing."""
+    from datetime import timedelta
+
+    config = _make_local_config()
+    config.last_run_at = datetime.now(timezone.utc) - timedelta(minutes=30)  # ran 30m ago
+    config.schedule_cron = "0 */6 * * *"  # every 6h — not due yet
+
+    mock_db_result_local = MagicMock()
+    mock_db_result_local.scalars.return_value.all.return_value = [config]
+
+    mock_db_result_cloud = MagicMock()
+    mock_db_result_cloud.scalars.return_value.all.return_value = []
+
+    mgr = _make_manager()
+
+    with patch("app.core.agent_runner.async_session") as mock_session_factory, \
+         patch("app.core.agent_runner.run_local_agent", new_callable=AsyncMock) as mock_run:
+        mock_ctx = AsyncMock()
+        mock_ctx.__aenter__ = AsyncMock(return_value=mock_ctx)
+        mock_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_ctx.execute = AsyncMock(
+            side_effect=[mock_db_result_local, mock_db_result_cloud]
+        )
+        mock_session_factory.return_value = mock_ctx
+
+        await trigger_pending_runs(_FREE_UID, "dev-001", mgr)
+
+    mock_run.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_trigger_pending_runs_device_id_filter():
+    """Local agents are only triggered for the matching device_id."""
+    # The DB query already filters by device_id, so we verify the SELECT
+    # includes the device_id filter by checking that a config bound to a
+    # different device is never dispatched.
+    #
+    # Since trigger_pending_runs queries with device_id == "dev-001",
+    # simulate the DB returning an empty list (as it would for a mismatch).
+    mock_db_result_local = MagicMock()
+    mock_db_result_local.scalars.return_value.all.return_value = []  # no match
+
+    mock_db_result_cloud = MagicMock()
+    mock_db_result_cloud.scalars.return_value.all.return_value = []
+
+    mgr = _make_manager(device_id="dev-001")
+
+    with patch("app.core.agent_runner.async_session") as mock_session_factory, \
+         patch("app.core.agent_runner.run_local_agent", new_callable=AsyncMock) as mock_run:
+        mock_ctx = AsyncMock()
+        mock_ctx.__aenter__ = AsyncMock(return_value=mock_ctx)
+        mock_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_ctx.execute = AsyncMock(
+            side_effect=[mock_db_result_local, mock_db_result_cloud]
+        )
+        mock_session_factory.return_value = mock_ctx
+
+        await trigger_pending_runs(_FREE_UID, "dev-001", mgr)
+
+    mock_run.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_trigger_pending_runs_dispatches_overdue():
+    """Overdue local agent triggers run_local_agent sequentially."""
+    config = _make_local_config()  # last_run_at=None → always overdue
+
+    mock_db_result_local = MagicMock()
+    mock_db_result_local.scalars.return_value.all.return_value = [config]
+
+    mock_db_result_cloud = MagicMock()
+    mock_db_result_cloud.scalars.return_value.all.return_value = []
+
+    mgr = _make_manager()
+
+    call_order: list[str] = []
+
+    async def _mock_run_local(user_id, cfg, run_log, device_mgr):
+        call_order.append("run_local")
+
+    with patch("app.core.agent_runner.async_session") as mock_session_factory, \
+         patch("app.core.agent_runner.run_local_agent", side_effect=_mock_run_local):
+        # First call: query configs. Subsequent calls: create run_log.
+        mock_query_ctx = AsyncMock()
+        mock_query_ctx.__aenter__ = AsyncMock(return_value=mock_query_ctx)
+        mock_query_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_query_ctx.execute = AsyncMock(
+            side_effect=[mock_db_result_local, mock_db_result_cloud]
+        )
+
+        run_log_obj = AgentRunLog(
+            id=str(uuid.uuid4()),
+            agent_id=config.id,
+            agent_type="local",
+            user_id=_FREE_UID,
+            status="running",
+            started_at=datetime.now(timezone.utc),
+        )
+        mock_insert_ctx = AsyncMock()
+        mock_insert_ctx.__aenter__ = AsyncMock(return_value=mock_insert_ctx)
+        mock_insert_ctx.__aexit__ = AsyncMock(return_value=False)
+        mock_insert_ctx.add = MagicMock()
+        mock_insert_ctx.commit = AsyncMock()
+        mock_insert_ctx.refresh = AsyncMock(side_effect=lambda obj: None)
+
+        mock_session_factory.side_effect = [mock_query_ctx, mock_insert_ctx]
+
+        await trigger_pending_runs(_FREE_UID, "dev-001", mgr)
+
+    assert call_order == ["run_local"]
+
+
+# ---------------------------------------------------------------------------
+# Integration: POST /agents/{id}/run
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _override_db(db_session):
+    """Route all get_session calls to the test SQLite session."""
+
+    async def _gen():
+        yield db_session
+
+    app.dependency_overrides[get_session] = _gen
+    yield
+    app.dependency_overrides.pop(get_session, None)
+
+
+@pytest.mark.asyncio
+async def test_trigger_run_unknown_agent(client):
+    """POST /agents/{id}/run returns 404 for unknown agent id."""
+    resp = client.post(
+        f"/api/v1/agents/{uuid.uuid4()}/run",
+        headers=auth_header("power"),
+    )
+    assert resp.status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_trigger_run_local_agent_creates_run_log(client, db_session):
+    """POST /agents/{id}/run creates a run log and dispatches a background task."""
+    # Create the local agent config in the DB.
+    config = LocalAgentConfig(
+        id=str(uuid.uuid4()),
+        user_id=TEST_USER_IDS["power"],
+        device_id="dev-001",
+        name="My Agent",
+        directory_paths=["/home/user/docs"],
+        data_types=["tasks"],
+        prompt_template="Extract tasks.",
+        file_extensions=[".txt"],
+        schedule_cron="0 */6 * * *",
+        enabled=True,
+    )
+    db_session.add(config)
+    await db_session.commit()
+
+    dispatched: list = []
+
+    async def _fake_run(user_id, cfg, run_log, device_mgr):
+        dispatched.append((user_id, cfg.id))
+
+    with patch("app.api.routes.agents.run_local_agent", new_callable=AsyncMock, side_effect=_fake_run), \
+         patch("app.api.routes.agents.run_cloud_agent", new_callable=AsyncMock), \
+         patch("asyncio.create_task") as mock_create_task:
+        resp = client.post(
+            f"/api/v1/agents/{config.id}/run",
+            headers=auth_header("power"),
+        )
+
+    assert resp.status_code == 202
+    data = resp.json()
+    assert data["agent_id"] == config.id
+    assert data["status"] == "running"
+    assert data["agent_type"] == "local"
+
+    # Verify create_task was called (dispatching background run).
+    mock_create_task.assert_called_once()

tests/test_device_ws.py

@@ -0,0 +1,362 @@
+"""Tests for Step 3.3: DeviceConnectionManager and device WS endpoint.
+
+Coverage:
+  Unit tests  — DeviceConnectionManager register/unregister/is_online/
+                get_ws/send_frame/pending-call round-trip/agent-data queue
+  Integration — /api/v1/ws/device endpoint via TestClient WebSocket:
+                auth rejection, happy-path connect, tool_result dispatch,
+                agent_data queue routing, agent_complete sentinel, disconnect
+                cleanup (AgentRunLog marked as error)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import pytest_asyncio
+
+from app.core.device_manager import DeviceConnection, DeviceConnectionManager
+from app.db import get_session
+from app.main import app
+from app.models import AgentRunLog
+from tests.conftest import TEST_USER_IDS, auth_header, make_jwt
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_FREE_UID = TEST_USER_IDS["free"]
+_PRO_UID = TEST_USER_IDS["pro"]
+
+
+def _device_hello(device_id: str = "dev-001", agent_ids: list[str] | None = None) -> str:
+    return json.dumps(
+        {"type": "device_hello", "device_id": device_id, "agent_ids": agent_ids or []}
+    )
+
+
+# ---------------------------------------------------------------------------
+# DB override (shared across integration tests)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture(autouse=True)
+def _override_db(db_session):
+    """Route all get_session calls to the test SQLite session."""
+
+    async def _gen():
+        yield db_session
+
+    app.dependency_overrides[get_session] = _gen
+    yield
+    app.dependency_overrides.pop(get_session, None)
+
+
+# ---------------------------------------------------------------------------
+# DeviceConnectionManager unit tests
+# ---------------------------------------------------------------------------
+
+@pytest.fixture()
+def manager() -> DeviceConnectionManager:
+    """Fresh manager instance for each test."""
+    return DeviceConnectionManager()
+
+
+@pytest.fixture()
+def mock_ws() -> MagicMock:
+    ws = MagicMock()
+    ws.send_text = AsyncMock()
+    return ws
+
+
+def test_manager_register_and_is_online(manager, mock_ws):
+    assert not manager.is_online("user1")
+    manager.register("user1", "dev-A", mock_ws)
+    assert manager.is_online("user1")
+    assert manager.is_online("user1", "dev-A")
+    assert not manager.is_online("user1", "dev-B")
+
+
+def test_manager_get_ws_returns_none_when_offline(manager):
+    assert manager.get_ws("no-such-user") is None
+
+
+def test_manager_unregister(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    assert manager.is_online("user1")
+    manager.unregister("user1")
+    assert not manager.is_online("user1")
+    assert manager.get_ws("user1") is None
+
+
+def test_manager_unregister_unknown_is_noop(manager):
+    # Must not raise.
+    manager.unregister("ghost")
+
+
+def test_manager_replace_connection_cancels_old_futures(manager):
+    ws_a = MagicMock()
+    ws_a.send_text = AsyncMock()
+    ws_b = MagicMock()
+    ws_b.send_text = AsyncMock()
+
+    # Create event loop context for Future.
+    loop = asyncio.new_event_loop()
+    try:
+        async def _run():
+            manager.register("user1", "dev-A", ws_a)
+            fut = manager.create_pending_call("user1", "call-1")
+            # Replace connection — old future should be cancelled.
+            manager.register("user1", "dev-B", ws_b)
+            assert fut.cancelled()
+
+        loop.run_until_complete(_run())
+    finally:
+        loop.close()
+
+
+@pytest.mark.asyncio
+async def test_manager_send_frame(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    await manager.send_frame("user1", {"type": "ping"})
+    mock_ws.send_text.assert_called_once_with(json.dumps({"type": "ping"}))
+
+
+@pytest.mark.asyncio
+async def test_manager_send_frame_raises_when_offline(manager):
+    with pytest.raises(RuntimeError, match="not connected"):
+        await manager.send_frame("ghost", {"type": "ping"})
+
+
+@pytest.mark.asyncio
+async def test_manager_pending_call_round_trip(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    fut = manager.create_pending_call("user1", "call-42")
+    result = {"type": "tool_result", "id": "call-42", "rows": [{"id": "row1"}]}
+    manager.resolve_pending_call("user1", "call-42", result)
+    assert fut.done()
+    assert await fut == result
+
+
+@pytest.mark.asyncio
+async def test_manager_resolve_unknown_call_is_noop(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    # Should not raise.
+    manager.resolve_pending_call("user1", "no-such-call", {})
+
+
+@pytest.mark.asyncio
+async def test_manager_unregister_cancels_pending_calls(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    fut = manager.create_pending_call("user1", "call-1")
+    manager.unregister("user1")
+    assert fut.cancelled()
+
+
+@pytest.mark.asyncio
+async def test_manager_agent_data_queue(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    q = manager.get_agent_data_queue("user1", "run-xyz")
+    # Put a frame and get it back.
+    frame = {"type": "agent_data", "run_id": "run-xyz", "files": []}
+    await q.put(frame)
+    assert await q.get() == frame
+
+
+@pytest.mark.asyncio
+async def test_manager_agent_data_queue_creates_once(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    q1 = manager.get_agent_data_queue("user1", "run-1")
+    q2 = manager.get_agent_data_queue("user1", "run-1")
+    assert q1 is q2
+
+
+@pytest.mark.asyncio
+async def test_manager_agent_data_queue_raises_when_offline(manager):
+    with pytest.raises(RuntimeError, match="not connected"):
+        manager.get_agent_data_queue("ghost", "run-1")
+
+
+@pytest.mark.asyncio
+async def test_manager_cleanup_agent_data_queue(manager, mock_ws):
+    manager.register("user1", "dev-A", mock_ws)
+    manager.get_agent_data_queue("user1", "run-1")
+    manager.cleanup_agent_data_queue("user1", "run-1")
+    # After cleanup a new queue is created (not the same object).
+    q_new = manager.get_agent_data_queue("user1", "run-1")
+    assert q_new is not None
+
+
+# ---------------------------------------------------------------------------
+# Integration tests — /api/v1/ws/device endpoint
+# ---------------------------------------------------------------------------
+
+def test_ws_device_rejects_without_token(client):
+    with pytest.raises(Exception):
+        # TestClient will raise or close when the server rejects.
+        with client.websocket_connect("/api/v1/ws/device") as ws:
+            ws.receive_text()
+
+
+def test_ws_device_rejects_invalid_token(client):
+    with pytest.raises(Exception):
+        with client.websocket_connect("/api/v1/ws/device?token=badtoken") as ws:
+            ws.receive_text()
+
+
+def test_ws_device_happy_path(client):
+    """Connect, send device_hello, receive ping, then close."""
+    token = make_jwt(tier="free")
+
+    # Patch the heartbeat sleep so the test doesn't block 30 s.
+    with patch("app.api.routes.device_ws._HEARTBEAT_INTERVAL", 0.01):
+        with client.websocket_connect(f"/api/v1/ws/device?token={token}") as ws:
+            ws.send_text(_device_hello("dev-001"))
+            # Next message from server should be a heartbeat ping (interval=0.01s).
+            msg = ws.receive_text()
+            data = json.loads(msg)
+            assert data["type"] == "ping"
+            # Close gracefully.
+            ws.close()
+
+
+def test_ws_device_invalid_first_frame_closes(client):
+    """Non-device_hello first frame should close the connection."""
+    token = make_jwt(tier="free")
+    with pytest.raises(Exception):
+        with client.websocket_connect(f"/api/v1/ws/device?token={token}") as ws:
+            ws.send_text(json.dumps({"type": "chat_request", "message": "hi"}))
+            ws.receive_text()  # server should close after bad frame
+
+
+def test_ws_device_tool_result_dispatched(client):
+    """tool_result frame is routed to the DeviceConnectionManager."""
+    token = make_jwt(tier="free")
+    user_id = TEST_USER_IDS["free"]
+
+    from app.core.device_manager import device_manager as dm
+
+    captured: list[dict] = []
+
+    original_resolve = dm.resolve_pending_call
+
+    def _spy(uid, call_id, result):
+        captured.append({"uid": uid, "call_id": call_id, "result": result})
+        original_resolve(uid, call_id, result)
+
+    with patch.object(dm, "resolve_pending_call", side_effect=_spy):
+        with patch("app.api.routes.device_ws._HEARTBEAT_INTERVAL", 9999):
+            with client.websocket_connect(f"/api/v1/ws/device?token={token}") as ws:
+                ws.send_text(_device_hello("dev-001"))
+                # Send a tool_result frame.
+                ws.send_text(
+                    json.dumps(
+                        {
+                            "type": "tool_result",
+                            "id": "call-123",
+                            "rows": [{"id": "task-1", "title": "Buy milk"}],
+                        }
+                    )
+                )
+                ws.close()
+
+    assert any(c["call_id"] == "call-123" for c in captured)
+
+
+def test_ws_device_agent_data_enqueued(client):
+    """agent_data frame is placed in the per-run queue by the message loop."""
+    from app.core.device_manager import device_manager as dm
+
+    token = make_jwt(tier="free")
+    user_id = TEST_USER_IDS["free"]
+
+    # Capture the queue object the message loop accesses.
+    captured_queue: list[asyncio.Queue] = []
+    original_get_queue = dm.get_agent_data_queue
+
+    def _spy_get_queue(uid, run_id):
+        q = original_get_queue(uid, run_id)
+        if not captured_queue:
+            captured_queue.append(q)
+        return q
+
+    with patch.object(dm, "get_agent_data_queue", side_effect=_spy_get_queue):
+        with patch("app.api.routes.device_ws._HEARTBEAT_INTERVAL", 9999):
+            with client.websocket_connect(f"/api/v1/ws/device?token={token}") as ws:
+                ws.send_text(_device_hello("dev-001"))
+                ws.send_text(
+                    json.dumps(
+                        {
+                            "type": "agent_data",
+                            "run_id": "run-XYZ",
+                            "files": [{"path": "/tmp/file.txt", "content": "hello"}],
+                        }
+                    )
+                )
+                ws.close()
+
+    # The queue should have received exactly one frame.
+    assert captured_queue, "queue was never accessed"
+    assert not captured_queue[0].empty()
+
+
+def test_ws_device_disconnect_marks_run_logs_as_error(client, db_session):
+    """On disconnect, _mark_runs_disconnected is called with the correct user_id."""
+    from app.api.routes import device_ws as _dws
+
+    token = make_jwt(tier="free")
+    user_id = TEST_USER_IDS["free"]
+
+    cleanup_calls: list[str] = []
+
+    async def _fake_cleanup(uid: str) -> None:
+        cleanup_calls.append(uid)
+
+    with patch.object(_dws, "_mark_runs_disconnected", side_effect=_fake_cleanup):
+        with patch("app.api.routes.device_ws._HEARTBEAT_INTERVAL", 9999):
+            with client.websocket_connect(f"/api/v1/ws/device?token={token}") as ws:
+                ws.send_text(_device_hello("dev-001"))
+                ws.close()
+
+    assert user_id in cleanup_calls
+
+
+@pytest.mark.asyncio
+async def test_mark_runs_disconnected_updates_db(db_session):
+    """_mark_runs_disconnected marks in-progress runs as error in the DB."""
+    from sqlalchemy import select
+
+    from app.api.routes.device_ws import _mark_runs_disconnected
+    from tests.conftest import _TestSessionLocal
+
+    user_id = TEST_USER_IDS["free"]
+
+    run_log = AgentRunLog(
+        id=str(uuid.uuid4()),
+        agent_id=str(uuid.uuid4()),
+        agent_type="local",
+        user_id=user_id,
+        status="running",
+        started_at=datetime.now(timezone.utc),
+    )
+    db_session.add(run_log)
+    await db_session.commit()
+
+    # Route the function to the same test-DB session factory.
+    with patch("app.api.routes.device_ws.async_session", _TestSessionLocal):
+        await _mark_runs_disconnected(user_id)
+
+    # Verify through the same session factory.
+    async with _TestSessionLocal() as s:
+        result = await s.execute(
+            select(AgentRunLog).where(AgentRunLog.id == run_log.id)
+        )
+        updated = result.scalar_one_or_none()
+
+    assert updated is not None
+    assert updated.status == "error"
+    assert updated.errors and "device disconnected" in updated.errors